refactor: 重构ext下的扩展工具

botpy/api.py

@@ -735,9 +735,9 @@ async def mute_member(
         self, guild_id: str, user_id: str, mute_end_timestamp: str = None, mute_seconds: str = None
     ) -> str:
         """
-        使频道中的所有成员禁言。
+        使频道中的指定成员禁言。
 
-        用于将频道的全体成员（非管理员）禁言。
+        用于将频道的指定成员（非管理员）禁言。
         需要使用的 token 对应的用户具备管理员权限。如果是机器人，要求被添加为管理员。
 
         Args:
@@ -1181,6 +1181,7 @@ async def get_pins(self, channel_id: str) -> pins_message.PinsMessage:
         )
         return await self._http.request(route)
 
+    # 帖子相关接口
     async def get_threads(self, channel_id: str) -> forum.ForumRsp:
         """
         该接口用于获取子频道下的帖子列表。

botpy/ext/__init__.py

@@ -2,4 +2,4 @@
 """
 这里放一些可用的工具
 方便开发者调用
-"""
+"""

botpy/ext/channel_jump_util.py → botpy/ext/channel_jump/__init__.py

@@ -1,70 +1,71 @@
-"""
-对子频道转跳进行操作(#name)
-注意:
-1、发送格式要求严格(#name )，自动添加的空格不能删除
-2、无法识别真假转跳
-3、当子频道重名时无法准确识别
-4、当提供子频道转跳字段时请弃用本模块
-"""
-
-__all__ = [
-    "get_channel_jump",
-    "get_channel_jump_strict",
-    "escape_channel_jump"
-]
-
-import re
-from typing import List, Dict
-
-from botpy import BotAPI
-from botpy.message import Message
-
-
-def get_channel_jump(text: str = None, message: Message = None) -> List[str]:
-    """
-    识别文本中的子频道转跳(粗略)
-    :param message: 消息对象
-    :param text: 文本，为空则message.content
-    :return: 子频道名称列表(不带#)
-    """
-    channel_jump_re = re.compile(r"#(.{1,12}?)(?= )")
-    return channel_jump_re.findall(message.content if text is None else text)
-
-
-async def get_channel_jump_strict(api: BotAPI, message: Message = None, text: str = None,
-                                  guild_id: str = None) -> Dict[str, str]:
-    """
-    识别文本中的子频道转跳(准确)
-    :param api: BotAPI
-    :param message: 消息对象
-    :param text: 文本，为空则message.content
-    :param guild_id: 频道id，为空则message.guild_id
-    :return: {子频道名称(不带#):子频道id} （去重）
-    """
-    channels = await api.get_channels(guild_id or message.guild_id)
-    text = message.content if text is None else text
-    jumps = {}
-
-    for channel in channels:
-        if "#%s " % channel["name"] in text:
-            jumps[channel["name"]] = channel["id"]
-
-    return jumps
-
-
-async def escape_channel_jump(api: BotAPI, message: Message = None, text: str = None, guild_id: str = None) -> str:
-    """
-    转义子频道转跳 (#name -> <#id>)
-    :param api: BotAPI
-    :param message: 消息对象
-    :param text: 文本，为空则message.content
-    :param guild_id: 频道id，为空则message.guild_id
-    :return: 转义后的文本
-    """
-    channels = await api.get_channels(guild_id or message.guild_id)
-    text = message.content if text is None else text
-
-    for channel in channels:
-        text = text.replace("#%s " % channel["name"], "<#%s> " % channel["id"])
-
-    return text
+# -*- coding: utf-8 -*-
+"""
+对子频道转跳进行操作(#name)
+注意:
+1、发送格式要求严格(#name )，自动添加的空格不能删除
+2、无法识别真假转跳
+3、当子频道重名时无法准确识别
+4、当提供子频道转跳字段时请弃用本模块
+"""
+
+__all__ = [
+    "get_channel_jump",
+    "get_channel_jump_strict",
+    "escape_channel_jump"
+]
+
+import re
+from typing import List, Dict
+
+from botpy import BotAPI
+from botpy.message import Message
+
+
+def get_channel_jump(text: str = None, message: Message = None) -> List[str]:
+    """
+    识别文本中的子频道转跳(粗略)
+    :param message: 消息对象
+    :param text: 文本，为空则message.content
+    :return: 子频道名称列表(不带#)
+    """
+    channel_jump_re = re.compile(r"#(.{1,12}?)(?= )")
+    return channel_jump_re.findall(message.content if text is None else text)
+
+
+async def get_channel_jump_strict(api: BotAPI, message: Message = None, text: str = None,
+                                  guild_id: str = None) -> Dict[str, str]:
+    """
+    识别文本中的子频道转跳(准确)
+    :param api: BotAPI
+    :param message: 消息对象
+    :param text: 文本，为空则message.content
+    :param guild_id: 频道id，为空则message.guild_id
+    :return: {子频道名称(不带#):子频道id} （去重）
+    """
+    channels = await api.get_channels(guild_id or message.guild_id)
+    text = message.content if text is None else text
+    jumps = {}
+
+    for channel in channels:
+        if "#%s " % channel["name"] in text:
+            jumps[channel["name"]] = channel["id"]
+
+    return jumps
+
+
+async def escape_channel_jump(api: BotAPI, message: Message = None, text: str = None, guild_id: str = None) -> str:
+    """
+    转义子频道转跳 (#name -> <#id>)
+    :param api: BotAPI
+    :param message: 消息对象
+    :param text: 文本，为空则message.content
+    :param guild_id: 频道id，为空则message.guild_id
+    :return: 转义后的文本
+    """
+    channels = await api.get_channels(guild_id or message.guild_id)
+    text = message.content if text is None else text
+
+    for channel in channels:
+        text = text.replace("#%s " % channel["name"], "<#%s> " % channel["id"])
+
+    return text

botpy/ext/cog_yaml/__init__.py

@@ -0,0 +1,16 @@
+# -*- coding: utf-8 -*-
+import yaml
+
+from typing import Dict, Any
+
+
+def read(yaml_path) -> Dict[str, Any]:
+    """
+    读取指定目录的yaml文件
+
+    :param yaml_path: 相对当前的yaml文件绝对路径
+    :return:
+    """
+    # 加上 ,encoding='utf-8'，处理配置文件中含中文出现乱码的情况。
+    with open(yaml_path, "r", encoding="utf-8") as f:
+        return yaml.safe_load(f)

botpy/ext/command_util.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 from functools import wraps
+from typing import Union
 
 from botpy import BotAPI
 from botpy.message import Message
@@ -8,10 +9,13 @@
 class Commands:
     """
     指令装饰器
+
+    Args:
+      name (Union[tuple, str]): 字符串元组或单个字符串。
     """
 
-    def __init__(self, commands: tuple or str):
-        self.commands = commands
+    def __init__(self, name: Union[tuple, str]):
+        self.commands = name
 
     def __call__(self, func):
         @wraps(func)
@@ -22,7 +26,7 @@ async def decorated(*args, **kwargs):
                 for command in self.commands:
                     if command in message.content:
                         # 分割指令后面的指令参数
-                        params = message.content.split(self.commands)[1].strip()
+                        params = message.content.split(command)[1].strip()
                         return await func(api=api, message=message, params=params)
             elif self.commands in message.content:
                 # 分割指令后面的指令参数

botpy/ext/converrt_color/__init__.py

@@ -0,0 +1,39 @@
+# -*- coding: utf-8 -*-
+from typing import Union
+
+
+def start(color: Union[tuple, str]) -> int:
+    """
+    将 RGB 值的元组或 HEX 值的字符串转换为单个整数
+
+    Args:
+      color (tuple or str): 输入RGB的三位tuple或HEX的sting颜色
+
+    Returns:
+      颜色的 RGB 值。
+    """
+    colors = []
+    if isinstance(color, tuple):
+        if len(color) == 3:
+            for items in color:
+                if not isinstance(items, int) or items not in range(256):
+                    raise TypeError("RGB颜色应为一个三位数的tuple，且当中每个数值都应该介乎于0和255之间，如(255,255,255)")
+                colors.append(items)
+        else:
+            raise TypeError("RGB颜色应为一个三位数的tuple，且当中每个数值都应该介乎于0和255之间，如(255,255,255)")
+    elif isinstance(color, str):
+        colour = color.replace("#", "")
+        if len(colour) == 6:
+            for items in (colour[:2], colour[2:4], colour[4:]):
+                try:
+                    items = int(items, 16)
+                except ValueError:
+                    raise TypeError("该HEX颜色不存在，请检查其颜色值是否准确")
+                if items not in range(256):
+                    raise TypeError("该HEX颜色不存在，请检查其颜色值是否准确")
+                colors.append(items)
+        else:
+            raise TypeError('HEX颜色应为一个 #加六位数字或字母 的string，如"#ffffff"')
+    else:
+        raise TypeError('颜色值应为RGB的三位tuple，如(255,255,255)；或HEX的sting颜色，如"#ffffff"')
+    return colors[0] + 256 * colors[1] + (256**2) * colors[2]

examples/demo_announce.py

@@ -1,12 +1,14 @@
+# -*- coding: utf-8 -*-
 import os
 
 import botpy
 from botpy import logging
+
 from botpy.message import Message
 from botpy.types.announce import AnnouncesType
-from botpy.ext.yaml_util import YamlUtil
+from botpy.ext.cog_yaml import read
 
-test_config = YamlUtil.read(os.path.join(os.path.dirname(__file__), "config.yaml"))
+test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
 
 _log = logging.get_logger()
 

examples/demo_api_permission.py

@@ -1,14 +1,14 @@
-#!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-import os.path
+import os
 
 import botpy
 from botpy import logging
+
 from botpy.message import Message
-from botpy.ext.yaml_util import YamlUtil
 from botpy.types.permission import APIPermissionDemandIdentify
+from botpy.ext.cog_yaml import read
 
-test_config = YamlUtil.read(os.path.join(os.path.dirname(__file__), "config.yaml"))
+test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
 
 _log = logging.get_logger()
 

examples/demo_at_reply.py

@@ -1,11 +1,13 @@
+# -*- coding: utf-8 -*-
 import os
 
 import botpy
 from botpy import logging
+
 from botpy.message import Message
-from botpy.ext.yaml_util import YamlUtil
+from botpy.ext.cog_yaml import read
 
-test_config = YamlUtil.read(os.path.join(os.path.dirname(__file__), "config.yaml"))
+test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
 
 _log = logging.get_logger()
 
@@ -15,6 +17,7 @@ async def on_ready(self):
         _log.info(f"robot 「{self.robot.name}」 on_ready!")
 
     async def on_at_message_create(self, message: Message):
+        _log.info(message.author.get("avatar"))
         await message.reply(content=f"机器人{self.robot.name}收到你的@消息了: {message.content}")
 
 

examples/demo_at_reply_ark.py

@@ -1,12 +1,14 @@
+# -*- coding: utf-8 -*-
 import os
 
 import botpy
 from botpy import logging
+
 from botpy.message import Message
 from botpy.types.message import Ark, ArkKv
-from botpy.ext.yaml_util import YamlUtil
+from botpy.ext.cog_yaml import read
 
-test_config = YamlUtil.read(os.path.join(os.path.dirname(__file__), "config.yaml"))
+test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
 
 _log = logging.get_logger()