ayaka.group

  1'''群组'''
  2from typing import List, Dict, TYPE_CHECKING
  3from pathlib import Path
  4from .storage import AyakaPath
  5from .config import ayaka_root_config
  6from .constant import app_list, group_list
  7from .cache import AyakaCacheCtrl
  8
  9if TYPE_CHECKING:
 10    from .ayaka import AyakaApp
 11
 12
 13def get_group(bot_id: int, group_id: int):
 14    '''获取对应的AyakaGroup对象，自动增加'''
 15    for group in group_list:
 16        if group.bot_id == bot_id and group.group_id == group_id:
 17            break
 18    else:
 19        group = AyakaGroup(bot_id, group_id)
 20    return group
 21
 22
 23class AyakaGroup:
 24    def __repr__(self) -> str:
 25        return f"AyakaGroup({self.bot_id}, {self.group_id}, {self.apps})"
 26
 27    def forbid_init(self):
 28        names = [
 29            "data", "groups",
 30            self.bot_id,
 31            self.group_id
 32        ]
 33        names = [str(name) for name in names]
 34        path = Path(*names)
 35        self.store_forbid = AyakaPath(path).json("forbid", [])
 36
 37    def forbid_load(self):
 38        return self.store_forbid.load()
 39
 40    def forbid_save(self, data):
 41        return self.store_forbid.save(data)
 42
 43    def __init__(self, bot_id: int, group_id: int) -> None:
 44        self.bot_id = bot_id
 45        self.group_id = group_id
 46        self.running_app: "AyakaApp" = None
 47
 48        # 读取forbit列表
 49        self.forbid_init()
 50        forbid_names = self.forbid_load()
 51
 52        # 添加app，并分配独立数据空间
 53        self.apps: List["AyakaApp"] = []
 54        self.cache_dict: Dict[str, AyakaCacheCtrl] = {}
 55        for app in app_list:
 56            if app.name not in forbid_names:
 57                self.apps.append(app)
 58                self.cache_dict[app.name] = AyakaCacheCtrl()
 59
 60        group_list.append(self)
 61
 62        if ayaka_root_config.debug:
 63            print(self)
 64
 65    @property
 66    def running_app_name(self):
 67        if self.running_app:
 68            return self.running_app.name
 69        return ""
 70
 71    def get_app(self, name: str):
 72        '''根据app名获取该group所启用的app，不存在则返回None'''
 73        for app in self.apps:
 74            if app.name == name:
 75                return app
 76
 77    def permit_app(self, name: str):
 78        '''启用指定app'''
 79        if self.get_app(name):
 80            return True
 81
 82        for app in app_list:
 83            if app.name == name:
 84                self.apps.append(app)
 85                # 从forbit列表移除
 86                app_names: list = self.forbid_load()
 87                if name in app_names:
 88                    app_names.remove(name)
 89                    self.forbid_save(app_names)
 90                return True
 91
 92    def forbid_app(self, name: str):
 93        '''禁用指定app'''
 94        if name == "ayaka_master":
 95            return
 96
 97        app = self.get_app(name)
 98        if not app:
 99            return
100
101        # 禁用正在运行的应用
102        if self.running_app_name == name:
103            self.running_app = None
104
105        # 移除
106        self.apps.remove(app)
107
108        # 添加到forbit列表
109        app_names: list = self.forbid_load()
110        if name not in app_names:
111            app_names.append(name)
112            self.forbid_save(app_names)
113        return True