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

添加注释，增加redis支持，添加了聊天记录构建支持

Browse Source
This commit is contained in:
K2cr2O1
2026-01-02 17:10:42 +08:00
parent 3163bbf8c1
commit 01b83803c1
24 changed files with 965 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
core/api/account.py
View File

@@ -1,78 +1,106 @@
"""
账号相关 API 模块
账号与状态相关 API 模块
该模块定义了 `AccountAPI` Mixin 类，提供了所有与机器人自身账号信息、
状态设置等相关的 OneBot v11 API 封装。
"""
import json
from typing import Dict, Any
from .base import BaseAPI
from models.objects import LoginInfo, VersionInfo, Status
from core.redis_manager import redis_client as redis_manager
class AccountAPI(BaseAPI):
"""
账号相关 API Mixin
`AccountAPI` Mixin 类，提供了所有与机器人账号、状态相关的 API 方法。
"""
async def get_login_info(self) -> LoginInfo:
async def get_login_info(self, no_cache: bool = False) -> LoginInfo:
"""
获取登录号信息
获取当前登录的机器人账号信息。
:return: 登录信息对象
Args:
no_cache (bool, optional): 是否不使用缓存，直接从服务器获取最新信息。Defaults to False.
Returns:
LoginInfo: 包含登录号 QQ 和昵称的 `LoginInfo` 数据对象。
"""
cache_key = f"neobot:cache:get_login_info:{self.self_id}"
if not no_cache:
cached_data = await redis_manager.get(cache_key)
if cached_data:
return LoginInfo(**json.loads(cached_data))
res = await self.call_api("get_login_info")
await redis_manager.set(cache_key, json.dumps(res), ex=3600) # 缓存 1 小时
return LoginInfo(**res)
async def get_version_info(self) -> VersionInfo:
"""
获取版本信息
获取 OneBot v11 实现的版本信息。
:return: 版本信息对象
Returns:
VersionInfo: 包含 OneBot 实现版本信息的 `VersionInfo` 数据对象。
"""
res = await self.call_api("get_version_info")
return VersionInfo(**res)
async def get_status(self) -> Status:
"""
获取状态
获取 OneBot v11 实现的状态信息。
:return: 状态对象
Returns:
Status: 包含 OneBot 状态信息的 `Status` 数据对象。
"""
res = await self.call_api("get_status")
return Status(**res)
async def bot_exit(self) -> Dict[str, Any]:
"""
退出机器人
让机器人进程退出（需要实现端支持）。
:return: API 响应结果
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
long_nick (str): 要设置的个性签名内容。
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
user_id (int): 目标用户的 QQ 号。
event_type (int): 事件类型。
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
face_id (int): 状态的表情 ID。
face_type (int): 状态的表情类型。
wording (str): 状态的描述文本。
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("set_diy_online_status", {
"face_id": face_id,
@@ -82,43 +110,55 @@ class AccountAPI(BaseAPI):
async def set_online_status(self, status_code: int) -> Dict[str, Any]:
"""
设置在线状态
设置在线状态（如在线、离开、摸鱼等）。
:param status_code: 状态码
:return: API 响应结果
Args:
status_code (int): 目标在线状态的状态码。
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
**kwargs: 个人资料的相关参数，具体字段请参考 OneBot v11 规范。
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
**kwargs: 头像的相关参数，具体字段请参考 OneBot v11 规范。
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("set_qq_avatar", kwargs)
async def get_clientkey(self) -> Dict[str, Any]:
"""
获取客户端密钥
获取客户端密钥（通常用于 QQ 登录相关操作）。
:return: API 响应结果
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("get_clientkey")
async def clean_cache(self) -> Dict[str, Any]:
"""
清理缓存
清理 OneBot v11 实现端的缓存。
:return: API 响应结果
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("clean_cache")

69
core/api/friend.py
View File

@@ -1,53 +1,86 @@
"""
好友相关 API 模块
好友与陌生人相关 API 模块
该模块定义了 `FriendAPI` Mixin 类，提供了所有与好友、陌生人信息
等相关的 OneBot v11 API 封装。
"""
import json
from typing import List, Dict, Any
from .base import BaseAPI
from models.objects import FriendInfo, StrangerInfo
from core.redis_manager import redis_client as redis_manager
class FriendAPI(BaseAPI):
"""
好友相关 API Mixin
`FriendAPI` Mixin 类，提供了所有与好友、陌生人操作相关的 API 方法。
"""
async def send_like(self, user_id: int, times: int = 1) -> Dict[str, Any]:
"""
发送点赞
向指定用户发送 "戳一戳" (点赞)。
:param user_id: 对方 QQ 号
:param times: 点赞次数
:return: API 响应结果
Args:
user_id (int): 目标用户的 QQ 号。
times (int, optional): 点赞次数，建议不超过 10 次。Defaults to 1.
Returns:
Dict[str, Any]: OneBot 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: 陌生人信息对象
Args:
user_id (int): 目标用户的 QQ 号。
no_cache (bool, optional): 是否不使用缓存，直接从服务器获取。Defaults to False.
Returns:
StrangerInfo: 包含陌生人信息的 `StrangerInfo` 数据对象。
"""
cache_key = f"neobot:cache:get_stranger_info:{user_id}"
if not no_cache:
cached_data = await redis_manager.get(cache_key)
if cached_data:
return StrangerInfo(**json.loads(cached_data))
res = await self.call_api("get_stranger_info", {"user_id": user_id, "no_cache": no_cache})
await redis_manager.set(cache_key, json.dumps(res), ex=3600) # 缓存 1 小时
return StrangerInfo(**res)
async def get_friend_list(self) -> List[FriendInfo]:
async def get_friend_list(self, no_cache: bool = False) -> List[FriendInfo]:
"""
获取好友列表
获取机器人账号的好友列表。
:return: 好友信息对象列表
Args:
no_cache (bool, optional): 是否不使用缓存，直接从服务器获取最新信息。Defaults to False.
Returns:
List[FriendInfo]: 包含所有好友信息的 `FriendInfo` 对象列表。
"""
cache_key = f"neobot:cache:get_friend_list:{self.self_id}"
if not no_cache:
cached_data = await redis_manager.get(cache_key)
if cached_data:
return [FriendInfo(**item) for item in json.loads(cached_data)]
res = await self.call_api("get_friend_list")
await redis_manager.set(cache_key, json.dumps(res), ex=3600) # 缓存 1 小时
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 响应结果
Args:
flag (str): 请求的标识，需要从 `request` 事件中获取。
approve (bool, optional): 是否同意该好友请求。Defaults to True.
remark (str, optional): 在同意请求时，为该好友设置的备注。Defaults to "".
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("set_friend_add_request", {"flag": flag, "approve": approve, "remark": remark})

228
core/api/group.py
View File

@@ -1,47 +1,64 @@
"""
群组相关 API 模块
该模块定义了 `GroupAPI` Mixin 类，提供了所有与群组管理、成员操作
等相关的 OneBot v11 API 封装。
"""
from typing import List, Dict, Any, Optional
import json
from core.redis_manager import redis_client as redis_manager
from .base import BaseAPI
from models.objects import GroupInfo, GroupMemberInfo, GroupHonorInfo
class GroupAPI(BaseAPI):
"""
群组相关 API Mixin
`GroupAPI` Mixin 类，提供了所有与群组操作相关的 API 方法。
"""
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 响应结果
Args:
group_id (int): 目标群组的群号。
user_id (int): 要踢出的成员的 QQ 号。
reject_add_request (bool, optional): 是否拒绝该用户此后的加群请求。Defaults to False.
Returns:
Dict[str, Any]: OneBot 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]:
async def set_group_ban(self, group_id: int, user_id: int, duration: int = 1800) -> Dict[str, Any]:
"""
群组单人禁言
禁言群组中的指定成员。
:param group_id: 群号
:param user_id: 要禁言的 QQ 号
:param duration: 禁言时长（秒），0 表示解除禁言
:return: API 响应结果
Args:
group_id (int): 目标群组的群号。
user_id (int): 要禁言的成员的 QQ 号。
duration (int, optional): 禁言时长，单位为秒。设置为 0 表示解除禁言。
Defaults to 1800 (30 分钟).
Returns:
Dict[str, Any]: OneBot 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]:
async def set_group_anonymous_ban(self, group_id: int, anonymous: Dict[str, Any] = None, duration: int = 1800, flag: str = None) -> Dict[str, Any]:
"""
群组匿名禁言
禁言群组中的匿名用户。
:param group_id: 群号
:param anonymous: 可选，要禁言的匿名用户对象（群消息事件的 anonymous 字段）
:param duration: 禁言时长（秒）
:param flag: 可选，要禁言的匿名用户的 flag（需从群消息事件的 anonymous 字段中获取）
:return: API 响应结果
Args:
group_id (int): 目标群组的群号。
anonymous (Dict[str, Any], optional): 要禁言的匿名用户对象，
可从群消息事件的 `anonymous` 字段中获取。Defaults to None.
duration (int, optional): 禁言时长，单位为秒。Defaults to 1800.
flag (str, optional): 要禁言的匿名用户的 flag 标识，
可从群消息事件的 `anonymous` 字段中获取。Defaults to None.
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
params = {"group_id": group_id, "duration": duration}
if anonymous:
@@ -52,139 +69,196 @@ class GroupAPI(BaseAPI):
async def set_group_whole_ban(self, group_id: int, enable: bool = True) -> Dict[str, Any]:
"""
群组全员禁言
开启或关闭群组全员禁言。
:param group_id: 群号
:param enable: 是否开启
:return: API 响应结果
Args:
group_id (int): 目标群组的群号。
enable (bool, optional): True 表示开启全员禁言，False 表示关闭。Defaults to True.
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
group_id (int): 目标群组的群号。
user_id (int): 目标成员的 QQ 号。
enable (bool, optional): True 表示设为管理员，False 表示取消管理员。Defaults to True.
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
group_id (int): 目标群组的群号。
enable (bool, optional): True 表示开启匿名，False 表示关闭。Defaults to True.
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
group_id (int): 目标群组的群号。
user_id (int): 目标成员的 QQ 号。
card (str, optional): 要设置的群名片内容。
传入空字符串 `""` 或 `None` 表示删除该成员的群名片。Defaults to "".
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
group_id (int): 目标群组的群号。
group_name (str): 新的群组名称。
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
group_id (int): 目标群组的群号。
is_dismiss (bool, optional): 是否解散群组。
仅当机器人是群主时，此项设为 True 才能解散群。Defaults to False.
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
group_id (int): 目标群组的群号。
user_id (int): 目标成员的 QQ 号。
special_title (str, optional): 专属头衔内容。
传入空字符串 `""` 或 `None` 表示删除头衔。Defaults to "".
duration (int, optional): 头衔有效期，单位为秒。-1 表示永久。Defaults to -1.
Returns:
Dict[str, Any]: OneBot 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: 群信息对象
Args:
group_id (int): 目标群组的群号。
no_cache (bool, optional): 是否不使用缓存，直接从服务器获取最新信息。Defaults to False.
Returns:
GroupInfo: 包含群组信息的 `GroupInfo` 数据对象。
"""
res = await self.call_api("get_group_info", {"group_id": group_id, "no_cache": no_cache})
cache_key = f"neobot:cache:get_group_info:{group_id}"
if not no_cache:
cached_data = await redis_manager.get(cache_key)
if cached_data:
return GroupInfo(**json.loads(cached_data))
res = await self.call_api("get_group_info", {"group_id": group_id})
await redis_manager.set(cache_key, json.dumps(res), ex=3600) # 缓存 1 小时
return GroupInfo(**res)
async def get_group_list(self) -> List[GroupInfo]:
"""
获取群列表
获取机器人加入的所有群组的列表。
:return: 群信息对象列表
Returns:
List[GroupInfo]: 包含所有群组信息的 `GroupInfo` 对象列表。
"""
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: 群成员信息对象
Args:
group_id (int): 目标群组的群号。
user_id (int): 目标成员的 QQ 号。
no_cache (bool, optional): 是否不使用缓存。Defaults to False.
Returns:
GroupMemberInfo: 包含群成员信息的 `GroupMemberInfo` 数据对象。
"""
res = await self.call_api("get_group_member_info", {"group_id": group_id, "user_id": user_id, "no_cache": no_cache})
cache_key = f"neobot:cache:get_group_member_info:{group_id}:{user_id}"
if not no_cache:
cached_data = await redis_manager.get(cache_key)
if cached_data:
return GroupMemberInfo(**json.loads(cached_data))
res = await self.call_api("get_group_member_info", {"group_id": group_id, "user_id": user_id})
await redis_manager.set(cache_key, json.dumps(res), ex=3600) # 缓存 1 小时
return GroupMemberInfo(**res)
async def get_group_member_list(self, group_id: int) -> List[GroupMemberInfo]:
"""
获取群成员列表
获取一个群组的所有成员列表。
:param group_id: 群号
:return: 群成员信息对象列表
Args:
group_id (int): 目标群组的群号。
Returns:
List[GroupMemberInfo]: 包含所有群成员信息的 `GroupMemberInfo` 对象列表。
"""
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: 群荣誉信息对象
Args:
group_id (int): 目标群组的群号。
type (str): 要获取的荣誉类型。
可选值: "talkative", "performer", "legend", "strong_newbie", "emotion" 等。
Returns:
GroupHonorInfo: 包含群荣誉信息的 `GroupHonorInfo` 数据对象。
"""
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 响应结果
Args:
flag (str): 请求的标识，需要从 `request` 事件中获取。
sub_type (str): 请求的子类型，`add` 或 `invite`，
需要与 `request` 事件中的 `sub_type` 字段相符。
approve (bool, optional): 是否同意请求或邀请。Defaults to True.
reason (str, optional): 拒绝加群的理由（仅在 `approve` 为 False 时有效）。Defaults to "".
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("set_group_add_request", {"flag": flag, "sub_type": sub_type, "approve": approve, "reason": reason})

136
core/api/message.py
View File

@@ -1,5 +1,8 @@
"""
消息相关 API 模块
该模块定义了 `MessageAPI` Mixin 类，提供了所有与消息发送、撤回、
转发等相关的 OneBot v11 API 封装。
"""
from typing import Union, List, Dict, Any, TYPE_CHECKING
from .base import BaseAPI
@@ -10,17 +13,22 @@ if TYPE_CHECKING:
class MessageAPI(BaseAPI):
"""
消息相关 API Mixin
`MessageAPI` Mixin 类，提供了所有与消息操作相关的 API 方法。
"""
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 响应结果
Args:
group_id (int): 目标群组的群号。
message (Union[str, MessageSegment, List[MessageSegment]]): 要发送的消息内容。
可以是纯文本字符串、单个消息段对象或消息段列表。
auto_escape (bool, optional): 仅当 `message` 为字符串时有效，
是否对消息内容进行 CQ 码转义。Defaults to False.
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api(
"send_group_msg", {"group_id": group_id, "message": self._process_message(message), "auto_escape": auto_escape}
@@ -28,12 +36,15 @@ class MessageAPI(BaseAPI):
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 响应结果
Args:
user_id (int): 目标用户的 QQ 号。
message (Union[str, MessageSegment, List[MessageSegment]]): 要发送的消息内容。
auto_escape (bool, optional): 是否对消息内容进行 CQ 码转义。Defaults to False.
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api(
"send_private_msg", {"user_id": user_id, "message": self._process_message(message), "auto_escape": auto_escape}
@@ -41,12 +52,18 @@ class MessageAPI(BaseAPI):
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 响应结果
该方法会根据传入的事件对象 `event` 自动判断是私聊还是群聊，
并调用相应的发送函数。如果事件是消息事件，则优先使用 `reply` 方法。
Args:
event (OneBotEvent): 触发该发送行为的事件对象。
message (Union[str, MessageSegment, List[MessageSegment]]): 要发送的消息内容。
auto_escape (bool, optional): 是否对消息内容进行 CQ 码转义。Defaults to False.
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
# 如果是消息事件，直接调用 reply
if hasattr(event, "reply"):
@@ -66,53 +83,98 @@ class MessageAPI(BaseAPI):
async def delete_msg(self, message_id: int) -> Dict[str, Any]:
"""
撤回消息
撤回一条消息。
:param message_id: 消息 ID
:return: API 响应结果
Args:
message_id (int): 要撤回的消息的 ID。
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
message_id (int): 要获取的消息的 ID。
Returns:
Dict[str, Any]: OneBot 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 响应结果
Args:
id (str): 合并转发消息的 ID。
Returns:
Dict[str, Any]: OneBot API 的响应数据，包含转发消息的节点列表。
"""
return await self.call_api("get_forward_msg", {"id": id})
async def send_group_forward_msg(self, group_id: int, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
发送群聊合并转发消息。
Args:
group_id (int): 目标群组的群号。
messages (List[Dict[str, Any]]): 消息节点列表。
推荐使用 `bot.build_forward_node` 来构建节点。
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("send_group_forward_msg", {"group_id": group_id, "messages": messages})
async def send_private_forward_msg(self, user_id: int, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
发送私聊合并转发消息。
Args:
user_id (int): 目标用户的 QQ 号。
messages (List[Dict[str, Any]]): 消息节点列表。
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("send_private_forward_msg", {"user_id": user_id, "messages": messages})
async def can_send_image(self) -> Dict[str, Any]:
"""
检查是否可以发送图片
检查当前机器人账号是否可以发送图片。
:return: API 响应结果
Returns:
Dict[str, Any]: OneBot API 的响应数据。
"""
return await self.call_api("can_send_image")
async def can_send_record(self) -> Dict[str, Any]:
"""
检查是否可以发送语音
检查当前机器人账号是否可以发送语音。
:return: API 响应结果
Returns:
Dict[str, Any]: OneBot 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 可接受的格式
内部方法：将消息内容处理成 OneBot API 可接受的格式。
:param message: 原始消息内容
:return: 处理后的消息内容
- `str` -> `str`
- `MessageSegment` -> `List[Dict]`
- `List[MessageSegment]` -> `List[Dict]`
Args:
message: 原始消息内容。
Returns:
处理后的消息内容。
"""
if isinstance(message, str):
return message
@@ -130,12 +192,16 @@ class MessageAPI(BaseAPI):
def _segment_to_dict(self, segment: "MessageSegment") -> Dict[str, Any]:
"""
将 MessageSegment 对象转换为字典
内部方法：将 `MessageSegment` 对象转换为字典。
:param segment: MessageSegment 对象
:return: 字典格式的消息段
Args:
segment (MessageSegment): 消息段对象。
Returns:
Dict[str, Any]: 符合 OneBot 规范的消息段字典。
"""
return {
"type": segment.type,
"data": segment.data
}