update code

utils/browser.py

@@ -13,14 +13,14 @@ _browser: Optional[Browser] = None
 
 
 async def init(**kwargs) -> Optional[Browser]:
-    try:
-        global _browser
-        browser = await async_playwright().start()
-        _browser = await browser.chromium.launch(**kwargs)
-        return _browser
-    except NotImplementedError:
-        logger.warning("win环境下 初始化playwright失败....请替换环境至linux")
-        return None
+    # try:
+    global _browser
+    browser = await async_playwright().start()
+    _browser = await browser.chromium.launch(**kwargs)
+    return _browser
+    # except NotImplementedError:
+    #     logger.warning("win环境下 初始化playwright失败....请替换环境至linux")
+    #     return None
 
 
 async def get_browser(**kwargs) -> Browser:

utils/image_utils.py

@@ -3,7 +3,7 @@ from PIL import Image, ImageFile, ImageDraw, ImageFont
 from imagehash import ImageHash
 from io import BytesIO
 from matplotlib import pyplot as plt
-from typing import Tuple, Optional
+from typing import Tuple, Optional, Union
 from pathlib import Path
 import cv2
 import base64
@@ -45,7 +45,7 @@ def get_img_hash(image_file: str) -> ImageHash:
     return hash_value
 
 
-def compressed_image(in_file: str, out_file: str = None, ratio: float = 0.9):
+def compressed_image(in_file: Union[str, Path], out_file: Union[str, Path] = None, ratio: float = 0.9):
     """
     说明：
         压缩图片
@@ -54,8 +54,11 @@ def compressed_image(in_file: str, out_file: str = None, ratio: float = 0.9):
         :param out_file: 压缩后输出的文件路径
         :param ratio: 压缩率，宽高 * 压缩率
     """
-    in_file = Path(IMAGE_PATH) / in_file
-    out_file = Path(IMAGE_PATH) / out_file if out_file else in_file
+    in_file = Path(IMAGE_PATH) / in_file if isinstance(in_file, str) else in_file
+    if out_file:
+        out_file = Path(IMAGE_PATH) / out_file if isinstance(out_file, str) else out_file
+    else:
+        out_file = in_file
     h, w, d = cv2.imread(str(in_file.absolute())).shape
     img = cv2.resize(cv2.imread(str(in_file.absolute())), (int(w * ratio), int(h * ratio)))
     cv2.imwrite(str(out_file.absolute()), img)

utils/static_data/__init__.py

@@ -0,0 +1,11 @@
+from typing import Optional
+from .group_manager import GroupManager
+from pathlib import Path
+from .data_source import init
+
+# 群权限
+group_manager: Optional[GroupManager] = GroupManager(
+    Path() / "data" / "manager" / "group_manager.json"
+)
+
+init(group_manager)

utils/static_data/data_class.py

@@ -0,0 +1,51 @@
+from typing import Union
+from pathlib import Path
+import ujson as json
+
+
+class StaticData:
+    """
+    静态数据共享类
+    """
+
+    def __init__(self, file: Path):
+        file.parent.mkdir(exist_ok=True, parents=True)
+        self.file = file
+        self.data = {}
+        if file.exists():
+            self.data: dict = json.load(open(file, "r", encoding="utf8"))
+
+    def set(self, key, value):
+        self.data[key] = value
+
+    def get(self, key):
+        return self.data.get(key)
+
+    def delete(self, key):
+        if self.data.get(key) is not None:
+            del self.data[key]
+
+    def save(self, path: Union[str, Path] = None):
+        path = path if path else self.file
+        with open(path, "w", encoding="utf8") as f:
+            json.dump(self.data, f, ensure_ascii=False, indent=4)
+
+    def reload(self):
+        if self.file.exists():
+            self.data: dict = json.load(open(self.file, "r", encoding="utf8"))
+
+    def is_exists(self):
+        return self.file.exists()
+
+    def is_empty(self):
+        return bool(len(self.data))
+
+    def __str__(self):
+        return str(self.data)
+
+    def __setitem__(self, key, value):
+        self.data[key] = value
+
+    def __getitem__(self, key):
+        return self.data[key]
+

utils/static_data/data_source.py

@@ -0,0 +1,27 @@
+from pathlib import Path
+import ujson as json
+from .group_manager import GroupManager
+
+
+def init(group_manager: GroupManager):
+    old_group_level_file = Path() / "data" / "manager" / "group_level.json"
+    old_plugin_list_file = Path() / "data" / "manager" / "plugin_list.json"
+    if old_group_level_file.exists():
+        data = json.load(open(old_group_level_file, 'r', encoding='utf8'))
+        for key in data.keys():
+            group = key
+            level = data[key]
+            group_manager.set_group_level(group, level)
+        old_group_level_file.unlink()
+        group_manager.save()
+
+    if old_plugin_list_file.exists():
+        data = json.load(open(old_plugin_list_file, 'r', encoding='utf8'))
+        for plugin in data.keys():
+            for group in data[plugin].keys():
+                if group == 'default' and not data[plugin]['default']:
+                    group_manager.block_plugin(plugin)
+                elif not data[plugin][group]:
+                    group_manager.block_plugin(plugin, group)
+        old_plugin_list_file.unlink()
+        group_manager.save()

utils/static_data/group_manager.py

@@ -0,0 +1,186 @@
+from typing import Optional
+from pathlib import Path
+from .data_class import StaticData
+
+
+class GroupManager(StaticData):
+    """
+    群权限 | 功能 | 聊天时间 管理器
+    """
+
+    def __init__(self, file: Path):
+        super().__init__(file)
+        if not self.data:
+            self.data = {"super": {"close_plugins": {}, "white_group_list": []}, "group_manager": {}}
+
+    def block_plugin(
+        self, plugin_cmd: str, group_id: Optional[str] = None, block_type: str = "all"
+    ):
+        """
+        说明：
+            锁定插件
+        参数：
+            :param plugin_cmd: 功能模块名
+            :param group_id: 群组，None时为超级用户禁用
+            :param block_type: 限制类型
+        """
+        self._set_plugin_status(plugin_cmd, "block", group_id, block_type)
+
+    def unblock_plugin(self, plugin_cmd: str, group_id: Optional[str] = None):
+        """
+        说明：
+            解锁插件
+        参数：
+            :param plugin_cmd: 功能模块名
+            :param group_id: 群组
+        """
+        self._set_plugin_status(plugin_cmd, "unblock", group_id)
+
+    def set_group_level(self, group_id: str, level: int):
+        """
+        说明：
+            设置群权限
+        参数：
+            :param group_id: 群组
+            :param level: 权限等级
+        """
+        if not self.data["group_manager"].get(group_id):
+            self._init_group(group_id)
+        self.data["group_manager"][group_id]["level"] = level
+        self.save()
+
+    def get_plugin_status(
+        self, plugin_cmd: str, group_id: Optional[str] = None, block_type: str = "all"
+    ) -> bool:
+        """
+        说明：
+            获取插件状态
+        参数：
+            :param plugin_cmd: 功能模块名
+            :param group_id: 群组
+            :param block_type: 限制类型
+        """
+        if group_id:
+            if not self.data["group_manager"].get(group_id):
+                self._init_group(group_id)
+                return True
+            if plugin_cmd in self.data["group_manager"][group_id]["close_plugins"]:
+                return False
+            return True
+        else:
+            if plugin_cmd in self.data["super"]["close_plugins"]:
+                if (
+                    self.data["super"]["close_plugins"][plugin_cmd] == "all"
+                    and block_type == "all"
+                ):
+                    return False
+                else:
+                    return (
+                        not self.data["super"]["close_plugins"][plugin_cmd]
+                        == block_type
+                    )
+            return True
+
+    def get_plugin_block_type(self, plugin_cmd: str) -> str:
+        """
+        说明：
+            获取功能限制类型
+        参数：
+            :param plugin_cmd: 模块名称
+        """
+        if plugin_cmd in self.data["super"]["close_plugins"]:
+            return self.data["super"]["close_plugins"][plugin_cmd]
+        return ""
+
+    def get_group_level(self, group_id: str) -> int:
+        """
+        说明：
+            获取群等级
+        参数：
+            :param group_id: 群号
+        """
+        if not self.data["group_manager"].get(group_id):
+            self._init_group(group_id)
+        return self.data["group_manager"][group_id]["level"]
+
+    def check_group_is_white(self, group_id: int) -> bool:
+        """
+        说明：
+            检测群聊是否在白名单
+        参数：
+            :param group_id: 群号
+        """
+        return group_id in self.data['super']['white_group_list']
+
+    def add_group_white_list(self, group_id: int):
+        """
+        说明：
+            将群聊加入白名单
+        参数：
+            :param group_id: 群号
+        """
+        if group_id not in self.data['super']['white_group_list']:
+            self.data['super']['white_group_list'].append(group_id)
+
+    def delete_group_white_list(self, group_id: int):
+        """
+        说明：
+            将群聊从白名单中删除
+        参数：
+            :param group_id: 群号
+        """
+        if group_id in self.data['super']['white_group_list']:
+            self.data['super']['white_group_list'].remove(group_id)
+
+    def _set_plugin_status(
+        self,
+        plugin_cmd: str,
+        status: str,
+        group_id: Optional[str],
+        block_type: str = "all",
+    ):
+        """
+        说明：
+            设置功能开关状态
+        参数：
+            :param plugin_cmd: 功能模块名
+            :param status: 功能状态
+            :param group_id: 群组
+            :param block_type: 限制类型
+        """
+        if group_id:
+            if not self.data["group_manager"].get(group_id):
+                self._init_group(group_id)
+            if status == "block":
+                if (
+                    plugin_cmd
+                    not in self.data["group_manager"][group_id]["close_plugins"]
+                ):
+                    self.data["group_manager"][group_id]["close_plugins"].append(
+                        plugin_cmd
+                    )
+            else:
+                if plugin_cmd in self.data["group_manager"][group_id]["close_plugins"]:
+                    self.data["group_manager"][group_id]["close_plugins"].remove(
+                        plugin_cmd
+                    )
+        else:
+            if status == "block":
+                if (
+                    plugin_cmd not in self.data["super"]["close_plugins"]
+                    or block_type != self.data["super"]["close_plugins"][plugin_cmd]
+                ):
+                    self.data["super"]["close_plugins"][plugin_cmd] = block_type
+            else:
+                if plugin_cmd in self.data["super"]["close_plugins"]:
+                    del self.data["super"]["close_plugins"][plugin_cmd]
+        self.save()
+
+    def _init_group(self, group_id: str):
+        """
+        说明：
+            初始化群数据
+        参数：
+            :param group_id: 群号
+        """
+        self.data["group_manager"][group_id] = {"level": 5, "close_plugins": []}