From 777343ac1a5dce7c1ed83597ae0b87605872ef56 Mon Sep 17 00:00:00 2001 From: yanyongyu Date: Wed, 26 Aug 2020 17:47:36 +0800 Subject: [PATCH] :label: Add typing for cqhttp api --- .editorconfig | 2 +- nonebot/adapters/cqhttp.py | 7 +- nonebot/adapters/cqhttp.pyi | 676 ++++++++++++++++++ nonebot/message.py | 5 +- tests/test_plugins/test_message.py | 2 +- .../test_plugins/test_package/test_command.py | 5 +- 6 files changed, 688 insertions(+), 9 deletions(-) create mode 100644 nonebot/adapters/cqhttp.pyi diff --git a/.editorconfig b/.editorconfig index bdeb8382..d4a3b991 100644 --- a/.editorconfig +++ b/.editorconfig @@ -33,5 +33,5 @@ trim_trailing_whitespace = false [{package.json,.travis.yml}] indent_size = 2 -[*.py] +[{*.py,*.pyi}] indent_size = 4 diff --git a/nonebot/adapters/cqhttp.py b/nonebot/adapters/cqhttp.py index bb9ec12d..3d071848 100644 --- a/nonebot/adapters/cqhttp.py +++ b/nonebot/adapters/cqhttp.py @@ -195,9 +195,10 @@ class Bot(BaseBot): @overrides(BaseBot) async def call_api(self, api: str, **data) -> Union[Any, NoReturn]: if "self_id" in data: - self_id = str(data.pop("self_id")) - bot = self.driver.bots[self_id] - return await bot.call_api(api, **data) + self_id = data.pop("self_id") + if self_id: + bot = self.driver.bots[str(self_id)] + return await bot.call_api(api, **data) if self.type == "websocket": seq = ResultStore.get_seq() diff --git a/nonebot/adapters/cqhttp.pyi b/nonebot/adapters/cqhttp.pyi new file mode 100644 index 00000000..d6ddd6d9 --- /dev/null +++ b/nonebot/adapters/cqhttp.pyi @@ -0,0 +1,676 @@ +from nonebot.adapters import BaseBot +from nonebot.typing import Any, Dict, List, Union, Message, Optional + + +class Bot(BaseBot): + + async def send_private_msg(self, + *, + user_id: int, + message: Union[str, Message], + auto_escape: bool = False, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 发送私聊消息。 + + :参数: + + * ``user_id``: 对方 QQ 号 + * ``message``: 要发送的内容 + * ``auto_escape``: 消息内容是否作为纯文本发送（即不解析 CQ 码），只在 ``message`` 字段是字符串时有效 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def send_group_msg(self, + *, + group_id: int, + message: Union[str, Message], + auto_escape: bool = False, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 发送群消息。 + + :参数: + + * ``group_id``: 群号 + * ``message``: 要发送的内容 + * ``auto_escape``: 消息内容是否作为纯文本发送（即不解析 CQ 码），只在 ``message`` 字段是字符串时有效 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def send_msg(self, + *, + message_type: Optional[str] = None, + user_id: Optional[int] = None, + group_id: Optional[int] = None, + message: Union[str, Message], + auto_escape: bool = False, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 发送消息。 + + :参数: + + * ``message_type``: 消息类型，支持 ``private``、``group``，分别对应私聊、群组、讨论组，如不传入，则根据传入的 ``*_id`` 参数判断 + * ``user_id``: 对方 QQ 号（消息类型为 ``private`` 时需要） + * ``group_id``: 群号（消息类型为 ``group`` 时需要） + * ``message``: 要发送的内容 + * ``auto_escape``: 消息内容是否作为纯文本发送（即不解析 CQ 码），只在 ``message`` 字段是字符串时有效 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def delete_msg(self, + *, + message_id: int, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 撤回消息。 + + :参数: + + * ``message_id``: 消息 ID + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_msg(self, + *, + message_id: int, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取消息。 + + :参数: + + * ``message_id``: 消息 ID + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_forward_msg(self, + *, + id: int, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 获取合并转发消息。 + + :参数: + + * ``id``: 合并转发 ID + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def send_like(self, + *, + user_id: int, + times: int = 1, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 发送好友赞。 + + :参数: + + * ``user_id``: 对方 QQ 号 + * ``times``: 赞的次数，每个好友每天最多 10 次 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_kick(self, + *, + group_id: int, + user_id: int, + reject_add_request: bool = False, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 群组踢人。 + + :参数: + + * ``group_id``: 群号 + * ``user_id``: 要踢的 QQ 号 + * ``reject_add_request``: 拒绝此人的加群请求 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_ban(self, + *, + group_id: int, + user_id: int, + duration: int = 30 * 60, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 群组单人禁言。 + + :参数: + + * ``group_id``: 群号 + * ``user_id``: 要禁言的 QQ 号 + * ``duration``: 禁言时长，单位秒，``0`` 表示取消禁言 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_anonymous_ban(self, + *, + group_id: int, + anonymous: Optional[Dict[str, + Any]] = None, + anonymous_flag: Optional[str] = None, + duration: int = 30 * 60, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 群组匿名用户禁言。 + + :参数: + + * ``group_id``: 群号 + * ``anonymous``: 可选，要禁言的匿名用户对象（群消息上报的 ``anonymous`` 字段） + * ``anonymous_flag``: 可选，要禁言的匿名用户的 flag（需从群消息上报的数据中获得） + * ``duration``: 禁言时长，单位秒，无法取消匿名用户禁言 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_whole_ban(self, + *, + group_id: int, + enable: bool = True, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 群组全员禁言。 + + :参数: + + * ``group_id``: 群号 + * ``enable``: 是否禁言 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_admin(self, + *, + group_id: int, + user_id: int, + enable: bool = True, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 群组设置管理员。 + + :参数: + + * ``group_id``: 群号 + * ``user_id``: 要设置管理员的 QQ 号 + * ``enable``: ``True`` 为设置，``False`` 为取消 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_anonymous(self, + *, + group_id: int, + enable: bool = True, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 群组匿名。 + + :参数: + + * ``group_id``: 群号 + * ``enable``: 是否允许匿名聊天 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_card(self, + *, + group_id: int, + user_id: int, + card: str = "", + self_id: Optional[int] = None) -> None: + """ + :说明: + + 设置群名片（群备注）。 + + :参数: + + * ``group_id``: 群号 + * ``user_id``: 要设置的 QQ 号 + * ``card``: 群名片内容，不填或空字符串表示删除群名片 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_name(self, + *, + group_id: int, + group_name: str, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 设置群名。 + + :参数: + + * ``group_id``: 群号 + * ``group_name``: 新群名 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_leave(self, + *, + group_id: int, + is_dismiss: bool = False, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 退出群组。 + + :参数: + + * ``group_id``: 群号 + * ``is_dismiss``: 是否解散，如果登录号是群主，则仅在此项为 True 时能够解散 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_special_title(self, + *, + group_id: int, + user_id: int, + special_title: str = "", + duration: int = -1, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 设置群组专属头衔。 + + :参数: + + * ``group_id``: 群号 + * ``user_id``: 要设置的 QQ 号 + * ``special_title``: 专属头衔，不填或空字符串表示删除专属头衔 + * ``duration``: 专属头衔有效期，单位秒，-1 表示永久，不过此项似乎没有效果，可能是只有某些特殊的时间长度有效，有待测试 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_friend_add_request(self, + *, + flag: str, + approve: bool = True, + remark: str = "", + self_id: Optional[int] = None) -> None: + """ + :说明: + + 处理加好友请求。 + + :参数: + + * ``flag``: 加好友请求的 flag（需从上报的数据中获得） + * ``approve``: 是否同意请求 + * ``remark``: 添加后的好友备注（仅在同意时有效） + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_group_add_request(self, + *, + flag: str, + sub_type: str, + approve: bool = True, + reason: str = "", + self_id: Optional[int] = None) -> None: + """ + :说明: + + 处理加群请求／邀请。 + + :参数: + + * ``flag``: 加群请求的 flag（需从上报的数据中获得） + * ``sub_type``: ``add`` 或 ``invite``，请求类型（需要和上报消息中的 ``sub_type`` 字段相符） + * ``approve``: 是否同意请求／邀请 + * ``reason``: 拒绝理由（仅在拒绝时有效） + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_login_info(self, + *, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取登录号信息。 + + :参数: + + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_stranger_info( + self, + *, + user_id: int, + no_cache: bool = False, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取陌生人信息。 + + :参数: + + * ``user_id``: QQ 号 + * ``no_cache``: 是否不使用缓存（使用缓存可能更新不及时，但响应更快） + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_friend_list(self, + *, + self_id: Optional[int] = None + ) -> List[Dict[str, Any]]: + """ + :说明: + + 获取好友列表。 + + :参数: + + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_group_info(self, + *, + group_id: int, + no_cache: bool = False, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取群信息。 + + :参数: + + * ``group_id``: 群号 + * ``no_cache``: 是否不使用缓存（使用缓存可能更新不及时，但响应更快） + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_group_list(self, + *, + self_id: Optional[int] = None + ) -> List[Dict[str, Any]]: + """ + :说明: + + 获取群列表。 + + :参数: + + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_group_member_info( + self, + *, + group_id: int, + user_id: int, + no_cache: bool = False, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取群成员信息。 + + :参数: + + * ``group_id``: 群号 + * ``user_id``: QQ 号 + * ``no_cache``: 是否不使用缓存（使用缓存可能更新不及时，但响应更快） + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_group_member_list( + self, + *, + group_id: int, + self_id: Optional[int] = None) -> List[Dict[str, Any]]: + """ + :说明: + + 获取群成员列表。 + + :参数: + + * ``group_id``: 群号 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_group_honor_info( + self, + *, + group_id: int, + type: str = "all", + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取群荣誉信息。 + + :参数: + + * ``group_id``: 群号 + * ``type``: 要获取的群荣誉类型，可传入 ``talkative`` ``performer`` ``legend`` ``strong_newbie`` ``emotion`` 以分别获取单个类型的群荣誉数据，或传入 ``all`` 获取所有数据 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_cookies(self, + *, + domain: str = "", + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取 Cookies。 + + :参数: + + * ``domain``: 需要获取 cookies 的域名 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_csrf_token(self, + *, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取 CSRF Token。 + + :参数: + + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_credentials(self, + *, + domain: str = "", + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取 QQ 相关接口凭证。 + + :参数: + + * ``domain``: 需要获取 cookies 的域名 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_record(self, + *, + file: str, + out_format: str, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取语音。 + + :参数: + + * ``file``: 收到的语音文件名（CQ 码的 ``file`` 参数），如 ``0B38145AA44505000B38145AA4450500.silk`` + * ``out_format``: 要转换到的格式，目前支持 ``mp3``、``amr``、``wma``、``m4a``、``spx``、``ogg``、``wav``、``flac`` + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_image(self, + *, + file: str, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取图片。 + + :参数: + + * ``file``: 收到的图片文件名（CQ 码的 ``file`` 参数），如 ``6B4DE3DFD1BD271E3297859D41C530F5.jpg`` + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def can_send_image(self, + *, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 检查是否可以发送图片。 + + :参数: + + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def can_send_record(self, + *, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 检查是否可以发送语音。 + + :参数: + + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_status(self, + *, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取插件运行状态。 + + :参数: + + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def get_version_info(self, + *, + self_id: Optional[int] = None) -> Dict[str, Any]: + """ + :说明: + + 获取版本信息。 + + :参数: + + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def set_restart(self, + *, + delay: int = 0, + self_id: Optional[int] = None) -> None: + """ + :说明: + + 重启 OneBot 实现。 + + :参数: + + * ``delay``: 要延迟的毫秒数，如果默认情况下无法重启，可以尝试设置延迟为 2000 左右 + * ``self_id``: 机器人 QQ 号 + """ + ... + + async def clean_cache(self, *, self_id: Optional[int] = None) -> None: + """ + :说明: + + 清理数据目录。 + + :参数: + + * ``self_id``: 机器人 QQ 号 + """ + ... diff --git a/nonebot/message.py b/nonebot/message.py index d405f6f8..e6488e20 100644 --- a/nonebot/message.py +++ b/nonebot/message.py @@ -34,6 +34,9 @@ async def _run_matcher(Matcher: Type[Matcher], bot: Bot, event: Event, logger.exception(e) return + # TODO: log matcher + logger.info(f"Event will be handled by {Matcher}") + matcher = Matcher() # TODO: BeforeMatcherRun try: @@ -102,7 +105,7 @@ async def handle_event(bot: Bot, event: Event): e_list = result.exceptions if StopPropagation in e_list: break_flag = True - logger.debug("Stop event propafation") + logger.debug("Stop event propagation") if ExpiredException in e_list: del matchers[priority][index - i] i += 1 diff --git a/tests/test_plugins/test_message.py b/tests/test_plugins/test_message.py index 51b4e198..76164bf7 100644 --- a/tests/test_plugins/test_message.py +++ b/tests/test_plugins/test_message.py @@ -12,7 +12,7 @@ test_message = on_message(state={"default": 1}) async def test_handler(bot: Bot, event: Event, state: dict): print("[*] Test Matcher Received:", event) state["event"] = event - await bot.send_private_msg(message="Received", user_id=event.user_id) + await bot.send(message="Received", event=event) @test_message.receive() diff --git a/tests/test_plugins/test_package/test_command.py b/tests/test_plugins/test_package/test_command.py index c2b5f334..8c811ddc 100644 --- a/tests/test_plugins/test_package/test_command.py +++ b/tests/test_plugins/test_package/test_command.py @@ -16,8 +16,7 @@ async def test_handler(bot: Bot, event: Event, state: dict): if args: state["help"] = args else: - await bot.send_private_msg(message="命令:\n1. test1\n2. test2", - user_id=event.user_id) + await bot.send(message="命令:\n1. test1\n2. test2", event=event) @test_command.got("help", prompt="你要帮助的命令是？") @@ -25,4 +24,4 @@ async def test_handler(bot: Bot, event: Event, state: dict): print("[!] Command 帮助:", state["help"]) if state["help"] not in ["test1", "test2"]: await test_command.reject(f"{state['help']} 不支持，请重新输入！") - await bot.send_private_msg(message=f"{state['help']} 帮助:\n...") + await bot.send(message=f"{state['help']} 帮助:\n...", event=event)