Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2026-09-30 09:10:01 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

优化manager代码

Browse Source
This commit is contained in:
HibiKier
2022-11-21 20:43:41 +08:00
parent 49e38862af
commit cd66b7f5b8
41 changed files with 1974 additions and 1760 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/manager/__init__.py
+3 -2
View File
@@ -14,6 +14,9 @@ from .requests_manager import RequestManager
from configs.path_config import DATA_PATH
# 管理员命令管理器
admin_manager = AdminManager()
# 群功能开关 | 群被动技能 | 群权限 管理
group_manager: Optional[GroupManager] = GroupManager(
DATA_PATH / "manager" / "group_manager.json"
@@ -62,5 +65,3 @@ requests_manager: Optional[RequestManager] = RequestManager(
DATA_PATH / "manager" / "requests_manager.json"
)
# 管理员命令管理器
admin_manager = AdminManager()
utils/manager/admin_manager.py
Executable → Regular
+38 -26
View File
@@ -1,5 +1,6 @@
from .data_class import StaticData
from typing import List, Optional
from .models import AdminSetting
from utils.manager.data_class import StaticData
from typing import List, Optional, Dict
class AdminManager(StaticData):
@@ -9,60 +10,71 @@ class AdminManager(StaticData):
def __init__(self):
super().__init__(None)
self._data: Dict[str, AdminSetting] = {}
def add_admin_plugin_settings(self, plugin: str, cmd: List[str], level: int):
"""
添加一个管理员命令
:param plugin: 模块
:param cmd: 别名
:param level: 等级
说明:
添加一个管理员命令
参数:
:param plugin: 模块
:param cmd: 别名
:param level: 等级
"""
self._data[plugin] = {
"level": level,
"cmd": cmd,
}
self._data[plugin] = AdminSetting(level=level, cmd=cmd)
def set_admin_level(self, plugin: str, level: int):
"""
设置管理员命令等级
:param plugin: 模块名
:param level: 权限等级
说明:
设置管理员命令等级
参数:
:param plugin: 模块名
:param level: 权限等级
"""
self._data[plugin]["level"] = level
if plugin in self._data.keys():
self._data[plugin].level = level
def remove_admin_plugin_settings(self, plugin: str):
"""
删除一个管理员命令
:param plugin: 模块名
说明:
删除一个管理员命令
参数:
:param plugin: 模块名
"""
if plugin in self._data.keys():
del self._data[plugin]
def check(self, plugin: str, level: int) -> bool:
"""
检查是否满足权限
:param plugin: 模块名
:param level: 权限等级
说明:
检查是否满足权限
参数:
:param plugin: 模块名
:param level: 权限等级
"""
if plugin in self._data.keys():
return level >= self._data[plugin]["level"]
return level >= self._data[plugin].level
return True
def get_plugin_level(self, plugin: str) -> int:
"""
获取插件等级
:param plugin: 模块名
说明:
获取插件等级
参数:
:param plugin: 模块名
"""
if plugin in self._data.keys():
return self._data[plugin]["level"]
return self._data[plugin].level
return 0
def get_plugin_module(self, cmd: str) -> Optional[str]:
"""
根据 cmd 获取功能 modules
:param cmd: 命令
说明:
根据 cmd 获取功能 modules
参数:
:param cmd: 命令
"""
for key in self._data.keys():
if self._data[key].get("cmd") and cmd in self._data[key]["cmd"]:
if self._data[key].cmd and cmd in self._data[key].cmd:
return key
return None
utils/manager/configs_manager.py
Executable → Regular
View File
utils/manager/data_class.py
+32 -18
View File
@@ -1,24 +1,29 @@
from typing import Union, Optional
from typing import Union, Optional, TypeVar, Generic, Dict, NoReturn, Any
from pathlib import Path
from ruamel.yaml import YAML
from ruamel import yaml
import ujson as json
import copy
from .models import *
_yaml = YAML(typ="safe")
class StaticData:
T = TypeVar("T", AdminSetting, BaseData, PluginBlock, PluginCd, PluginCount, PluginSetting, Plugin)
class StaticData(Generic[T]):
"""
静态数据共享类
"""
def __init__(self, file: Optional[Path]):
def __init__(self, file: Optional[Path], load_file: bool = True):
self._data: dict = {}
if file:
file.parent.mkdir(exist_ok=True, parents=True)
self.file = file
if file.exists():
if file.exists() and load_file:
with open(file, "r", encoding="utf8") as f:
if file.name.endswith("json"):
try:
@@ -29,30 +34,39 @@ class StaticData:
elif file.name.endswith("yaml"):
self._data = _yaml.load(f)
def set(self, key, value):
def set(self, key, value) -> NoReturn:
self._data[key] = value
self.save()
def set_module_data(self, module, key, value, auto_save: bool = True):
def set_module_data(self, module, key, value, auto_save: bool = True) -> NoReturn:
if module in self._data.keys():
self._data[module][key] = value
if auto_save:
self.save()
def get(self, key):
def get(self, key) -> T:
return self._data.get(key)
def keys(self):
def keys(self) -> List[str]:
return self._data.keys()
def delete(self, key):
def delete(self, key) -> NoReturn:
if self._data.get(key) is not None:
del self._data[key]
def get_data(self) -> dict:
def get_data(self) -> Dict[str, T]:
return copy.deepcopy(self._data)
def save(self, path: Union[str, Path] = None):
def dict(self) -> Dict[str, Any]:
temp = {}
for k, v in self._data.items():
try:
temp[k] = v.dict()
except AttributeError:
temp[k] = copy.deepcopy(v)
return temp
def save(self, path: Union[str, Path] = None) -> NoReturn:
path = path or self.file
if isinstance(path, str):
path = Path(path)
@@ -61,26 +75,26 @@ class StaticData:
if path.name.endswith("yaml"):
yaml.dump(self._data, f, indent=2, Dumper=yaml.RoundTripDumper, allow_unicode=True)
else:
json.dump(self._data, f, ensure_ascii=False, indent=4)
json.dump(self.dict(), f, ensure_ascii=False, indent=4)
def reload(self):
def reload(self) -> NoReturn:
if self.file.exists():
if self.file.name.endswith("json"):
self._data: dict = json.load(open(self.file, "r", encoding="utf8"))
elif self.file.name.endswith("yaml"):
self._data: dict = _yaml.load(open(self.file, "r", encoding="utf8"))
def is_exists(self):
def is_exists(self) -> bool:
return self.file.exists()
def is_empty(self):
def is_empty(self) -> bool:
return bool(len(self._data))
def __str__(self):
def __str__(self) -> str:
return str(self._data)
def __setitem__(self, key, value):
def __setitem__(self, key, value) -> NoReturn:
self._data[key] = value
def __getitem__(self, key):
def __getitem__(self, key) -> T:
return self._data[key]
utils/manager/group_manager.py
Executable → Regular
+352 -378
View File
@@ -1,378 +1,352 @@
from typing import Optional, List, Union, Dict
from pathlib import Path
from .data_class import StaticData
from utils.utils import get_matchers, get_bot
from configs.config import Config
import nonebot
Config.add_plugin_config(
"group_manager", "DEFAULT_GROUP_LEVEL", 5, help_="默认群权限", default_value=5
)
Config.add_plugin_config(
"group_manager", "DEFAULT_GROUP_BOT_STATUS", True, help_="默认进群总开关状态", default_value=True
)
class GroupManager(StaticData):
"""
群权限 | 功能 | 总开关 | 聊天时间 管理器
"""
def __init__(self, file: Path):
super().__init__(file)
if not self._data:
self._data = {
"super": {"white_group_list": []},
"group_manager": {},
}
self._task = {}
def block_plugin(self, module: str, group_id: int):
"""
说明:
锁定插件
参数:
:param module: 功能模块名
:param group_id: 群组，None时为超级用户禁用
"""
self._set_plugin_status(module, "block", group_id)
def unblock_plugin(self, module: str, group_id: int):
"""
说明:
解锁插件
参数:
:param module: 功能模块名
:param group_id: 群组
"""
self._set_plugin_status(module, "unblock", group_id)
def turn_on_group_bot_status(self, group_id: int):
"""
说明:
开启群bot开关
参数:
:param group_id: 群号
"""
self._set_group_bot_status(group_id, True)
def shutdown_group_bot_status(self, group_id: int):
"""
说明:
关闭群bot开关
参数:
:param group_id: 群号
"""
self._set_group_bot_status(group_id, False)
def check_group_bot_status(self, group_id: int) -> bool:
"""
说明:
检查群聊bot总开关状态
参数:
:param group_id: 说明
"""
group_id = str(group_id)
if not self._data["group_manager"].get(group_id):
self._init_group(group_id)
if self._data["group_manager"][group_id].get("status") is None:
default_group_bot_status = Config.get_config("group_manager", "DEFAULT_GROUP_BOT_STATUS")
if default_group_bot_status:
default_group_bot_status = True
self._data["group_manager"][group_id]["status"] = default_group_bot_status
return self._data["group_manager"][group_id]["status"]
def set_group_level(self, group_id: int, level: int):
"""
说明:
设置群权限
参数:
:param group_id: 群组
:param level: 权限等级
"""
group_id = str(group_id)
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, module: str, group_id: int) -> bool:
"""
说明:
获取插件状态
参数:
:param module: 功能模块名
:param group_id: 群组
"""
group_id = str(group_id) if group_id else group_id
if not self._data["group_manager"].get(group_id):
self._init_group(group_id)
return True
if module in self._data["group_manager"][group_id]["close_plugins"]:
return False
return True
def get_group_level(self, group_id: int) -> int:
"""
说明:
获取群等级
参数:
:param group_id: 群号
"""
group_id = str(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 get_group_white_list(self) -> List[str]:
"""
说明:
获取所有群白名单
"""
return self._data["super"]["white_group_list"]
def delete_group(self, group_id: int):
"""
说明:
删除群配置
参数:
:param group_id: 群号
"""
if group_id in self._data["group_manager"]:
del self._data["group_manager"][str(group_id)]
if group_id in self._data["super"]["white_group_list"]:
self._data["super"]["white_group_list"].remove(group_id)
self.save()
async def open_group_task(self, group_id: int, task: str):
"""
说明:
开启群被动技能
参数:
:param group_id: 群号
:param task: 被动技能名称
"""
await self._set_group_task_status(group_id, task, True)
async def close_group_task(self, group_id: int, task: str):
"""
说明:
关闭群被动技能
参数:
:param group_id: 群号
:param task: 被动技能名称
"""
await self._set_group_task_status(group_id, task, False)
async def check_group_task_status(self, group_id: int, task: str) -> bool:
"""
说明:
查看群被动技能状态
参数:
:param group_id: 群号
:param task: 被动技能名称
"""
group_id = str(group_id)
if (
not self._data["group_manager"][group_id].get("group_task_status")
or self._data["group_manager"][group_id]["group_task_status"].get(task)
is None
):
await self.init_group_task(group_id)
return self._data["group_manager"][group_id]["group_task_status"][task]
def get_task_data(self) -> Dict[str, str]:
"""
说明:
获取所有被动任务
"""
return self._task
async def group_task_status(self, group_id: int) -> str:
"""
说明:
查看群被全部动技能状态
参数:
:param group_id: 群号
"""
x = "[群被动技能]:\n"
group_id = str(group_id)
if not self._data["group_manager"][group_id].get("group_task_status"):
await self.init_group_task(group_id)
for key in self._data["group_manager"][group_id]["group_task_status"].keys():
x += f'{self._task[key]}：{"√" if await self.check_group_task_status(int(group_id), key) else "×"}\n'
return x[:-1]
async def _set_group_task_status(self, group_id: int, task: str, status: bool):
"""
说明:
管理群被动技能状态
参数:
:param group_id: 群号
:param task: 被动技能
:param status: 状态
"""
group_id = str(group_id)
if not self._data["group_manager"].get(group_id):
self._init_group(group_id)
if (
not self._data["group_manager"][group_id].get("group_task_status")
or self._data["group_manager"][group_id]["group_task_status"].get(task)
is None
):
await self.init_group_task(group_id)
self._data["group_manager"][group_id]["group_task_status"][task] = status
self.save()
async def init_group_task(self, group_id: Optional[Union[int, str]] = None):
"""
说明:
初始化群聊 被动技能 状态
"""
if not self._task:
_m = []
for matcher in get_matchers():
if matcher.plugin_name not in _m:
_m.append(matcher.plugin_name)
_plugin = nonebot.plugin.get_plugin(matcher.plugin_name)
try:
_module = _plugin.module
plugin_task = _module.__getattribute__("__plugin_task__")
for key in plugin_task.keys():
if key in self._task.keys():
raise ValueError(f"plugin_task：{key} 已存在！")
self._task[key] = plugin_task[key]
except AttributeError:
pass
bot = get_bot()
if bot or group_id:
if group_id:
_group_list = [group_id]
else:
_group_list = [x["group_id"] for x in await bot.get_group_list()]
for group_id in _group_list:
group_id = str(group_id)
if not self._data["group_manager"].get(group_id):
self._init_group(group_id)
if not self._data["group_manager"][group_id].get("group_task_status"):
self._data["group_manager"][group_id]["group_task_status"] = {}
for task in self._task:
if (
self._data["group_manager"][group_id]["group_task_status"].get(
task
)
is None
):
self._data["group_manager"][group_id]["group_task_status"][
task
] = Config.get_config('_task', f'DEFAULT_{task}', default=True)
for task in list(
self._data["group_manager"][group_id]["group_task_status"]
):
if task not in self._task:
del self._data["group_manager"][group_id]["group_task_status"][
task
]
self.save()
def _set_plugin_status(
self,
module: str,
status: str,
group_id: int,
):
"""
说明:
设置功能开关状态
参数:
:param module: 功能模块名
:param status: 功能状态
:param group_id: 群组
"""
group_id = str(group_id) if group_id else group_id
if not self._data["group_manager"].get(group_id):
self._init_group(group_id)
if status == "block":
if module not in self._data["group_manager"][group_id]["close_plugins"]:
self._data["group_manager"][group_id]["close_plugins"].append(module)
else:
if module in self._data["group_manager"][group_id]["close_plugins"]:
self._data["group_manager"][group_id]["close_plugins"].remove(module)
self.save()
def _init_group(self, group_id: str):
"""
说明:
初始化群数据
参数:
:param group_id: 群号
"""
default_group_level = Config.get_config("group_manager", "DEFAULT_GROUP_LEVEL")
if default_group_level is None:
default_group_level = 5
default_group_bot_status = Config.get_config("group_manager", "DEFAULT_GROUP_BOT_STATUS")
if default_group_bot_status:
default_group_bot_status = True
if not self._data["group_manager"].get(group_id):
self._data["group_manager"][group_id] = {
"level": default_group_level,
"status": default_group_bot_status,
"close_plugins": [],
"group_task_status": {},
}
def _set_group_bot_status(self, group_id: Union[int, str], status: bool):
"""
说明:
设置群聊bot总开关
参数:
:param group_id: 群号
:param status: 开关状态
"""
group_id = str(group_id)
if not self._data["group_manager"].get(group_id):
self._init_group(group_id)
self._data["group_manager"][group_id]["status"] = status
self.save()
def get_super_old_data(self) -> Optional[dict]:
"""
说明:
获取旧数据，平时使用请不要调用
"""
if self._data["super"].get("close_plugins"):
_x = self._data["super"].get("close_plugins")
del self._data["super"]["close_plugins"]
return _x
return None
from typing import List, Union, Dict, Callable
from pathlib import Path
from .models import BaseData, BaseGroup
from utils.manager.data_class import StaticData
from utils.utils import get_matchers, is_number
from configs.config import Config
import nonebot
import ujson as json
Config.add_plugin_config(
"group_manager", "DEFAULT_GROUP_LEVEL", 5, help_="默认群权限", default_value=5
)
Config.add_plugin_config(
"group_manager", "DEFAULT_GROUP_BOT_STATUS", True, help_="默认进群总开关状态", default_value=True
)
def init_group(func: Callable):
"""
说明:
初始化群数据
参数:
:param func: func
"""
def wrapper(*args, **kwargs):
self = args[0]
group_id = list(filter(lambda x: is_number(x), args[1:]))[0]
if self and group_id and not self._data.group_manager.get(str(group_id)):
self._data.group_manager[str(group_id)] = BaseGroup()
return func(*args, **kwargs)
return wrapper
def init_task(func: Callable):
"""
说明:
初始化群被动
参数:
:param func: func
"""
def wrapper(*args, **kwargs):
self = args[0]
group_id = str(args[1])
task = args[2] if len(args) > 1 else None
if group_id and task and self._data.group_manager[group_id].group_task_status.get(task) is None:
for task in self._data.task:
if self._data.group_manager[group_id].group_task_status.get(task) is None:
self._data.group_manager[group_id].group_task_status[task] = Config.get_config('_task', f'DEFAULT_{task}', default=True)
for task in list(self._data.group_manager[group_id].group_task_status):
if task not in self._data.task:
del self._data.group_manager[group_id].group_task_status[task]
self.save()
return func(*args, **kwargs)
return wrapper
class GroupManager(StaticData):
"""
群权限 | 功能 | 总开关 | 聊天时间 管理器
"""
def __init__(self, file: Path):
super().__init__(file, False)
self._data = BaseData.parse_file(file) if file.exists() else BaseData()
def block_plugin(self, module: str, group_id: int, is_save: bool = True):
"""
说明:
锁定插件
参数:
:param module: 功能模块名
:param group_id: 群组，None时为超级用户禁用
:param is_save: 是否保存文件
"""
self._set_plugin_status(module, "block", group_id, is_save)
def unblock_plugin(self, module: str, group_id: int, is_save: bool = True):
"""
说明:
解锁插件
参数:
:param module: 功能模块名
:param group_id: 群组
:param is_save: 是否保存文件
"""
self._set_plugin_status(module, "unblock", group_id, is_save)
def turn_on_group_bot_status(self, group_id: int):
"""
说明:
开启群bot开关
参数:
:param group_id: 群号
"""
self._set_group_bot_status(group_id, True)
def shutdown_group_bot_status(self, group_id: int):
"""
说明:
关闭群bot开关
参数:
:param group_id: 群号
"""
self._set_group_bot_status(group_id, False)
@init_group
def check_group_bot_status(self, group_id: int) -> bool:
"""
说明:
检查群聊bot总开关状态
参数:
:param group_id: 说明
"""
return self._data.group_manager[str(group_id)].status
@init_group
def set_group_level(self, group_id: int, level: int):
"""
说明:
设置群权限
参数:
:param group_id: 群组
:param level: 权限等级
"""
self._data.group_manager[str(group_id)].level = level
self.save()
@init_group
def get_plugin_status(self, module: str, group_id: int) -> bool:
"""
说明:
获取插件状态
参数:
:param module: 功能模块名
:param group_id: 群组
"""
return module not in self._data.group_manager[str(group_id)].close_plugins
@init_group
def get_group_level(self, group_id: int) -> int:
"""
说明:
获取群等级
参数:
:param group_id: 群号
"""
return self._data.group_manager[str(group_id)].level
def check_group_is_white(self, group_id: int) -> bool:
"""
说明:
检测群聊是否在白名单
参数:
:param group_id: 群号
"""
return group_id in self._data.white_group
def add_group_white_list(self, group_id: int):
"""
说明:
将群聊加入白名单
参数:
:param group_id: 群号
"""
if group_id not in self._data.white_group:
self._data.white_group.append(group_id)
def delete_group_white_list(self, group_id: int):
"""
说明:
将群聊从白名单中删除
参数:
:param group_id: 群号
"""
if group_id in self._data.white_group:
self._data.white_group.remove(group_id)
def get_group_white_list(self) -> List[int]:
"""
说明:
获取所有群白名单
"""
return self._data.white_group
def load_task(self):
"""
说明:
加载被动技能
"""
for matcher in get_matchers(True):
_plugin = nonebot.plugin.get_plugin(matcher.plugin_name)
try:
_module = _plugin.module
plugin_task = _module.__getattribute__("__plugin_task__")
for key in plugin_task.keys():
if key in self._data.task.keys():
raise ValueError(f"plugin_task：{key} 已存在！")
self._data.task[key] = plugin_task[key]
except AttributeError:
pass
@init_group
def delete_group(self, group_id: int):
"""
说明:
删除群配置
参数:
:param group_id: 群号
"""
if group_id in self._data.white_group:
self._data.white_group.remove(group_id)
self.save()
def open_group_task(self, group_id: int, task: str):
"""
说明:
开启群被动技能
参数:
:param group_id: 群号
:param task: 被动技能名称
"""
self._set_group_group_task_status(group_id, task, True)
def close_group_task(self, group_id: int, task: str):
"""
说明:
关闭群被动技能
参数:
:param group_id: 群号
:param task: 被动技能名称
"""
self._set_group_group_task_status(group_id, task, False)
@init_task
def check_group_task_status(self, group_id: int, task: str) -> bool:
"""
说明:
查看群被动技能状态
参数:
:param group_id: 群号
:param task: 被动技能名称
"""
return self._data.group_manager[str(group_id)].group_task_status.get(task, False)
def get_task_data(self) -> Dict[str, str]:
"""
说明:
获取所有被动任务
"""
return self._data.task
@init_task
def group_group_task_status(self, group_id: int) -> str:
"""
说明:
查看群被全部动技能状态
参数:
:param group_id: 群号
"""
x = "[群被动技能]:\n"
group_id = str(group_id)
for key in self._data.group_manager[group_id].group_task_status.keys():
x += f'{self._data.task[key]}：{"√" if self.check_group_task_status(int(group_id), key) else "×"}\n'
return x[:-1]
@init_task
@init_group
def _set_group_group_task_status(self, group_id: int, task: str, status: bool):
"""
说明:
管理群被动技能状态
参数:
:param group_id: 群号
:param task: 被动技能
:param status: 状态
"""
self._data.group_manager[str(group_id)].group_task_status[task] = status
self.save()
@init_group
def _set_plugin_status(
self,
module: str,
status: str,
group_id: int,
is_save: bool
):
"""
说明:
设置功能开关状态
参数:
:param module: 功能模块名
:param status: 功能状态
:param group_id: 群组
:param is_save: 是否保存
"""
group_id = str(group_id) if group_id else group_id
if status == "block":
if module not in self._data.group_manager[group_id].close_plugins:
self._data.group_manager[group_id].close_plugins.append(module)
else:
if module in self._data.group_manager[group_id].close_plugins:
self._data.group_manager[group_id].close_plugins.remove(module)
if is_save:
self.save()
@init_group
def _set_group_bot_status(self, group_id: Union[int, str], status: bool):
"""
说明:
设置群聊bot总开关
参数:
:param group_id: 群号
:param status: 开关状态
"""
self._data.group_manager[str(group_id)].status = status
self.save()
def reload(self):
if self.file.exists():
t = self._data.task
self._data = BaseData.parse_file(self.file)
self._data.task = t
def save(self, path: Union[str, Path] = None):
"""
说明:
保存文件
参数:
:param path: 路径文件
"""
path = path or self.file
if isinstance(path, str):
path = Path(path)
if path:
dict_data = self._data.dict()
del dict_data["task"]
with open(path, "w", encoding="utf8") as f:
json.dump(dict_data, f, ensure_ascii=False, indent=4)
# def get_super_old_data(self) -> Optional[dict]:
# """
# 说明:
# 获取旧数据，平时使用请不要调用
# """
# if self._data["super"].get("close_plugins"):
# _x = self._data["super"].get("close_plugins")
# del self._data["super"]["close_plugins"]
# return _x
# return None
utils/manager/models.py
+95
View File
@@ -0,0 +1,95 @@
from typing import List, Optional, Dict, Literal, Tuple, Union
from pydantic import BaseModel
from configs.config import Config
class AdminSetting(BaseModel):
"""
管理员设置
"""
level: int
cmd: Optional[List[str]]
class BaseGroup(BaseModel):
"""
基础群聊信息
"""
level: int = Config.get_config("group_manager", "DEFAULT_GROUP_LEVEL") # 群等级
status: bool = Config.get_config(
"group_manager", "DEFAULT_GROUP_BOT_STATUS"
) # 总开关状态
close_plugins: List[str] = [] # 已关闭插件
group_task_status: Dict[str, bool] = {} # 被动状态
class BaseData(BaseModel):
"""
群基本信息
"""
white_group: List[int] = [] # 白名单
group_manager: Dict[str, BaseGroup] = {} # 群组管理
task: Dict[str, str] = {} # 被动任务 【英文：中文】
class PluginBlock(BaseModel):
"""
插件阻断
"""
status: bool # 限制状态
check_type: Literal["private", "group", "all"] # 检查类型
limit_type: Literal["user", "group"] # 监听对象
rst: Optional[str] # 阻断时回复
class PluginCd(BaseModel):
"""
插件阻断
"""
cd: int # cd
status: bool # 限制状态
check_type: Literal["private", "group", "all"] # 检查类型
limit_type: Literal["user", "group"] # 监听对象
rst: Optional[str] # 阻断时回复
class PluginCount(BaseModel):
"""
插件阻断
"""
max_count: int # 次数
status: bool # 限制状态
limit_type: Literal["user", "group"] # 监听对象
rst: Optional[str] # 阻断时回复
class PluginSetting(BaseModel):
"""
插件设置
"""
cmd: List[str] = [] # 命令 或 命令别名
default_status: bool = True # 默认开关状态
level: int = 5 # 功能权限等级
limit_superuser: bool = False # 功能状态是否限制超级用户
plugin_type: Tuple[Union[str, int], ...] = ("normal",) # 插件类型
cost_gold: int = 0 # 需要消费的金币
class Plugin(BaseModel):
"""
插件数据
"""
plugin_name: str # 模块名
status: Optional[bool] = True # 开关状态
error: Optional[bool] = False # 是否加载报错
block_type: Optional[str] = None # 关闭类型
author: Optional[str] = None # 作者
version: Optional[int] = None # 版本
utils/manager/none_plugin_count_manager.py
Executable → Regular
+47 -47
View File
@@ -1,47 +1,47 @@
from .data_class import StaticData
from typing import Optional
from pathlib import Path
class NonePluginCountManager(StaticData):
"""
插件加载容忍管理器，当连续 max_count 次插件加载，视为删除插件，清楚数据
"""
def __init__(self, file: Optional[Path], max_count: int = 5):
"""
:param file: 存储路径
:param max_count: 容忍最大次数
"""
super().__init__(file)
if not self._data:
self._data = {}
self._max_count = max_count
def add_count(self, module: str, count: int = 1):
"""
添加次数
:param module: 模块
:param count: 次数，无特殊情况均为 1
"""
if module not in self._data.keys():
self._data[module] = count
else:
self._data[module] += count
def reset(self, module: str):
"""
重置次数
:param module: 模块
"""
if module in self._data.keys():
self._data[module] = 0
def check(self, module: str):
"""
检查容忍次数是否到达最大值
:param module: 模块
"""
if module in self._data.keys():
return self._data[module] >= self._max_count
return False
from utils.manager.data_class import StaticData
from typing import Optional
from pathlib import Path
class NonePluginCountManager(StaticData):
"""
插件加载容忍管理器，当连续 max_count 次插件加载，视为删除插件，清楚数据
"""
def __init__(self, file: Optional[Path], max_count: int = 5):
"""
:param file: 存储路径
:param max_count: 容忍最大次数
"""
super().__init__(file)
if not self._data:
self._data = {}
self._max_count = max_count
def add_count(self, module: str, count: int = 1):
"""
添加次数
:param module: 模块
:param count: 次数，无特殊情况均为 1
"""
if module not in self._data.keys():
self._data[module] = count
else:
self._data[module] += count
def reset(self, module: str):
"""
重置次数
:param module: 模块
"""
if module in self._data.keys():
self._data[module] = 0
def check(self, module: str):
"""
检查容忍次数是否到达最大值
:param module: 模块
"""
if module in self._data.keys():
return self._data[module] >= self._max_count
return False
utils/manager/plugins2block_manager.py
Executable → Regular
+103 -50
View File
@@ -1,11 +1,12 @@
from typing import Optional, Dict
from .data_class import StaticData
from typing import Optional, Dict, Literal, Union
from utils.manager.data_class import StaticData
from services.log import logger
from utils.utils import UserBlockLimiter
from pathlib import Path
from ruamel.yaml import YAML
from ruamel import yaml
from .models import PluginBlock
yaml = YAML(typ="safe")
_yaml = yaml.YAML(typ="safe")
class Plugins2blockManager(StaticData):
@@ -14,53 +15,45 @@ class Plugins2blockManager(StaticData):
"""
def __init__(self, file: Path):
self.file = file
super().__init__(None)
super().__init__(file, False)
self._block_limiter: Dict[str, UserBlockLimiter] = {}
if file.exists():
with open(file, "r", encoding="utf8") as f:
self._data = yaml.load(f)
if "PluginBlockLimit" in self._data.keys():
self._data = self._data["PluginBlockLimit"] or {}
self.__load_file()
def add_block_limit(
self,
plugin: str,
*,
status: Optional[bool] = True,
check_type: Optional[str] = "all",
limit_type: Optional[str] = "user",
check_type: Literal["private", "group", "all"] = "all",
limit_type: Literal["user", "group"] = "user",
rst: Optional[str] = None,
**kwargs # 用于接收额外实参
):
"""
添加插件调用 block 限制
:param plugin: 插件模块名称
:param status: 默认开关状态
:param check_type: 检查类型 'private'/'group'/'all'，限制私聊/群聊/全部
:param limit_type: 限制类型 监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
:param rst: 回复的话，为空则不回复
说明:
添加插件调用 block 限制
参数:
:param plugin: 插件模块名称
:param status: 默认开关状态
:param check_type: 检查类型 'private'/'group'/'all'，限制私聊/群聊/全部
:param limit_type: 限制类型 监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
:param rst: 回复的话，为空则不回复
"""
status = status or True
check_type = check_type or "all"
limit_type = limit_type or "user"
if check_type not in ["all", "group", "private"]:
raise ValueError(
f"{plugin} 添加block限制错误，‘check_type‘ 必须为 'private'/'group'/'all'"
)
if limit_type not in ["user", "group"]:
raise ValueError(f"{plugin} 添加block限制错误，‘limit_type‘ 必须为 'user'/'group'")
self._data[plugin] = {
"status": status,
"check_type": check_type,
"limit_type": limit_type,
"rst": rst,
}
self._data[plugin] = PluginBlock(
status=status, check_type=check_type, limit_type=limit_type, rst=rst
)
def get_plugin_block_data(self, plugin: str) -> Optional[dict]:
def get_plugin_block_data(self, plugin: str) -> Optional[PluginBlock]:
"""
获取插件block数据
:param plugin: 模块名
说明:
获取插件block数据
参数:
:param plugin: 模块名
"""
if self.check_plugin_block_status(plugin):
return self._data[plugin]
@@ -68,16 +61,20 @@ class Plugins2blockManager(StaticData):
def check_plugin_block_status(self, plugin: str) -> bool:
"""
检测插件是否有 block
:param plugin: 模块名
说明:
检测插件是否有 block
参数:
:param plugin: 模块名
"""
return plugin in self._data.keys() and self._data[plugin]["status"]
return plugin in self._data.keys() and self._data[plugin].status
def check(self, id_: int, plugin: str) -> bool:
"""
检查 block
:param plugin: 模块名
:param id_: 限制 id
说明:
检查 block
参数:
:param id_: 限制 id
:param plugin: 模块名
"""
if self._block_limiter.get(plugin):
return self._block_limiter[plugin].check(id_)
@@ -85,26 +82,30 @@ class Plugins2blockManager(StaticData):
def set_true(self, id_: int, plugin: str):
"""
对插件 block
:param plugin: 模块名
:param id_: 限制 id
说明:
对插件 block
参数:
:param id_: 限制 id
:param plugin: 模块名
"""
if self._block_limiter.get(plugin):
self._block_limiter[plugin].set_true(id_)
def set_false(self, id_: int, plugin: str):
"""
对插件 unblock
:param plugin: 模块名
:param id_: 限制 id
说明:
对插件 unblock
参数:
:param plugin: 模块名
:param id_: 限制 id
"""
if self._block_limiter.get(plugin):
self._block_limiter[plugin].set_false(id_)
def reload_block_limit(self):
"""
加载 block 限制器
:return:
说明:
加载 block 限制器
"""
for plugin in self._data:
if self.check_plugin_block_status(plugin):
@@ -113,10 +114,62 @@ class Plugins2blockManager(StaticData):
def reload(self):
"""
重载本地数据
说明:
重载本地数据
"""
self.__load_file()
self.reload_block_limit()
def save(self, path: Union[str, Path] = None):
"""
说明:
保存文件
参数:
:param path: 文件路径
"""
path = path or self.file
if isinstance(path, str):
path = Path(path)
if path:
with open(path, "w", encoding="utf8") as f:
yaml.dump(
{"PluginBlockLimit": self.dict()},
f,
indent=2,
Dumper=yaml.RoundTripDumper,
allow_unicode=True,
)
_data = yaml.round_trip_load(open(path, encoding="utf8"))
_data["PluginBlockLimit"].yaml_set_start_comment(
"""# 用户调用阻塞
# 即 当用户调用此功能还未结束时
# 用发送消息阻止用户重复调用此命令直到该命令结束
# key：模块名称
# status：此限制的开关状态
# check_type：'private'/'group'/'all'，限制私聊/群聊/全部
# limit_type：监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
# 示例：'user'：阻塞用户，'group'：阻塞群聊
# rst：回复的话，可以添加[at]，[uname]，[nickname]来对应艾特，用户群名称，昵称系统昵称
# rst 为 "" 或 None 时则不回复
# rst示例："[uname]你冲的太快了，[nickname]先生，请稍后再冲[at]"
# rst回复："老色批你冲的太快了，欧尼酱先生，请稍后再冲@老色批"
# 用户昵称↑ 昵称系统的昵称↑ 艾特用户↑""",
indent=2,
)
with open(path, "w", encoding="utf8") as wf:
yaml.round_trip_dump(
_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True
)
def __load_file(self):
"""
说明:
读取配置文件
"""
self._data: Dict[str, PluginBlock] = {}
if self.file.exists():
with open(self.file, "r", encoding="utf8") as f:
self._data: dict = yaml.load(f)
self._data = self._data["PluginBlockLimit"]
self.reload_block_limit()
temp = yaml.round_trip_load(f)
if "PluginBlockLimit" in temp.keys():
for k, v in temp["PluginBlockLimit"].items():
self._data[k] = PluginBlock.parse_obj(v)
utils/manager/plugins2cd_manager.py
Executable → Regular
+106 -61
View File
@@ -1,11 +1,12 @@
from typing import Optional, Dict
from .data_class import StaticData
from typing import Optional, Dict, Literal, Union
from utils.manager.data_class import StaticData
from utils.utils import FreqLimiter
from services.log import logger
from pathlib import Path
from ruamel.yaml import YAML
from ruamel import yaml
from .models import PluginCd
yaml = YAML(typ="safe")
_yaml = yaml.YAML(typ="safe")
class Plugins2cdManager(StaticData):
@@ -14,16 +15,9 @@ class Plugins2cdManager(StaticData):
"""
def __init__(self, file: Path):
self.file = file
super().__init__(None)
super().__init__(file, False)
self._freq_limiter: Dict[str, FreqLimiter] = {}
if file.exists():
with open(file, "r", encoding="utf8") as f:
self._data = yaml.load(f)
if "PluginCdLimit" in self._data.keys():
self._data = (
self._data["PluginCdLimit"] if self._data["PluginCdLimit"] else {}
)
self.__load_file()
def add_cd_limit(
self,
@@ -31,42 +25,35 @@ class Plugins2cdManager(StaticData):
*,
cd: Optional[int] = 5,
status: Optional[bool] = True,
check_type: Optional[str] = "all",
limit_type: Optional[str] = "user",
check_type: Literal["private", "group", "all"] = "all",
limit_type: Literal["user", "group"] = "user",
rst: Optional[str] = None,
**kwargs # 用于接收额外实参
):
"""
添加插件调用 cd 限制
:param plugin: 插件模块名称
:param cd: cd 时长
:param status: 默认开关状态
:param check_type: 检查类型 'private'/'group'/'all'，限制私聊/群聊/全部
:param limit_type: 限制类型 监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
:param rst: 回复的话，为空则不回复
说明:
添加插件调用 cd 限制
参数:
:param plugin: 插件模块名称
:param cd: cd 时长
:param status: 默认开关状态
:param check_type: 检查类型 'private'/'group'/'all'，限制私聊/群聊/全部
:param limit_type: 限制类型 监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
:param rst: 回复的话，为空则不回复
"""
cd = cd or 5
status = status or True
check_type = check_type or "all"
limit_type = limit_type or "user"
if check_type not in ["all", "group", "private"]:
raise ValueError(
f"{plugin} 添加cd限制错误，‘check_type‘ 必须为 'private'/'group'/'all'"
)
if limit_type not in ["user", "group"]:
raise ValueError(f"{plugin} 添加cd限制错误，‘limit_type‘ 必须为 'user'/'group'")
self._data[plugin] = {
"cd": cd,
"status": status,
"check_type": check_type,
"limit_type": limit_type,
"rst": rst,
}
self._data[plugin] = PluginCd(cd=cd, status=status, check_type=check_type, limit_type=limit_type, rst=rst)
def get_plugin_cd_data(self, plugin: str) -> Optional[dict]:
def get_plugin_cd_data(self, plugin: str) -> Optional[PluginCd]:
"""
获取插件cd数据
:param plugin: 模块名
说明:
获取插件cd数据
参数:
:param plugin: 模块名
"""
if self.check_plugin_cd_status(plugin):
return self._data[plugin]
@@ -74,20 +61,24 @@ class Plugins2cdManager(StaticData):
def check_plugin_cd_status(self, plugin: str) -> bool:
"""
检测插件是否有 cd
:param plugin: 模块名
说明:
检测插件是否有 cd
参数:
:param plugin: 模块名
"""
return (
plugin in self._data.keys()
and self._data[plugin]["cd"] > 0
and self._data[plugin]["status"]
and self._data[plugin].cd > 0
and self._data[plugin].status
)
def check(self, plugin: str, id_: int) -> bool:
"""
检查 cd
:param plugin: 模块名
:param id_: 限制 id
说明:
检查 cd
参数:
:param plugin: 模块名
:param id_: 限制 id
"""
if self._freq_limiter.get(plugin):
return self._freq_limiter[plugin].check(id_)
@@ -95,42 +86,96 @@ class Plugins2cdManager(StaticData):
def start_cd(self, plugin: str, id_: int, cd: int = 0):
"""
开始cd
:param plugin: 模块名
:param id_: cd 限制类型
:param cd: cd 时长
:return:
说明:
开始cd
参数:
:param plugin: 模块名
:param id_: cd 限制类型
:param cd: cd 时长
"""
if self._freq_limiter.get(plugin):
self._freq_limiter[plugin].start_cd(id_, cd)
def get_plugin_data(self, plugin: str) -> dict:
def get_plugin_data(self, plugin: str) -> Optional[PluginCd]:
"""
获取单个模块限制数据
:param plugin: 模块名
说明:
获取单个模块限制数据
参数:
:param plugin: 模块名
"""
if self._data.get(plugin) is not None:
if self._data.get(plugin):
return self._data.get(plugin)
return {}
def reload_cd_limit(self):
"""
加载 cd 限制器
:return:
说明:
加载 cd 限制器
"""
for plugin in self._data:
if self.check_plugin_cd_status(plugin):
self._freq_limiter[plugin] = FreqLimiter(
self.get_plugin_cd_data(plugin)["cd"]
self.get_plugin_cd_data(plugin).cd
)
logger.info(f"已成功加载 {len(self._freq_limiter)} 个Cd限制.")
def reload(self):
"""
重载本地数据
说明:
重载本地数据
"""
self.__load_file()
self.reload_cd_limit()
def save(self, path: Union[str, Path] = None):
"""
说明:
保存文件
参数:
:param path: 文件路径
"""
path = path or self.file
if isinstance(path, str):
path = Path(path)
if path:
with open(path, "w", encoding="utf8") as f:
yaml.dump(
{"PluginCdLimit": self.dict()},
f,
indent=2,
Dumper=yaml.RoundTripDumper,
allow_unicode=True,
)
_data = yaml.round_trip_load(open(path, encoding="utf8"))
_data["PluginCdLimit"].yaml_set_start_comment(
"""# 需要cd的功能
# 自定义的功能需要cd也可以在此配置
# key：模块名称
# cd：cd 时长（秒）
# status：此限制的开关状态
# check_type：'private'/'group'/'all'，限制私聊/群聊/全部
# limit_type：监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
# 示例：'user'：用户N秒内触发1次，'group'：群N秒内触发1次
# rst：回复的话，可以添加[at]，[uname]，[nickname]来对应艾特，用户群名称，昵称系统昵称
# rst 为 "" 或 None 时则不回复
# rst示例："[uname]你冲的太快了，[nickname]先生，请稍后再冲[at]"
# rst回复："老色批你冲的太快了，欧尼酱先生，请稍后再冲@老色批"
# 用户昵称↑ 昵称系统的昵称↑ 艾特用户↑""",
indent=2,
)
with open(path, "w", encoding="utf8") as wf:
yaml.round_trip_dump(
_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True
)
def __load_file(self):
"""
说明:
读取配置文件
"""
self._data: Dict[str, PluginCd] = {}
if self.file.exists():
with open(self.file, "r", encoding="utf8") as f:
self._data: dict = yaml.load(f)
self._data = self._data["PluginCdLimit"]
self.reload_cd_limit()
temp = _yaml.load(f)
if "PluginCdLimit" in temp.keys():
for k, v in temp["PluginCdLimit"].items():
self._data[k] = PluginCd.parse_obj(v)
utils/manager/plugins2count_manager.py
Executable → Regular
+101 -55
View File
@@ -1,11 +1,12 @@
from typing import Optional, Dict
from .data_class import StaticData
from typing import Optional, Dict, Literal, Union
from utils.manager.data_class import StaticData
from utils.utils import DailyNumberLimiter
from services.log import logger
from pathlib import Path
from ruamel.yaml import YAML
from ruamel import yaml
from .models import PluginCount
yaml = YAML(typ="safe")
_yaml = yaml.YAML(typ="safe")
class Plugins2countManager(StaticData):
@@ -14,16 +15,9 @@ class Plugins2countManager(StaticData):
"""
def __init__(self, file: Path):
self.file = file
super().__init__(None)
super().__init__(file, False)
self._daily_limiter: Dict[str, DailyNumberLimiter] = {}
if file.exists():
with open(file, "r", encoding="utf8") as f:
self._data = yaml.load(f)
if "PluginCountLimit" in self._data.keys():
self._data = (
self._data["PluginCountLimit"] if self._data["PluginCountLimit"] else {}
)
self.__load_file()
def add_count_limit(
self,
@@ -31,64 +25,64 @@ class Plugins2countManager(StaticData):
*,
max_count: int = 5,
status: Optional[bool] = True,
limit_type: Optional[str] = "user",
limit_type: Literal["user", "group"] = "user",
rst: Optional[str] = None,
**kwargs # 用于接收额外实参
):
"""
添加插件调用 次数 限制
:param plugin: 插件模块名称
:param max_count: 最大次数限制
:param status: 默认开关状态
:param limit_type: 限制类型 监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
:param rst: 回复的话，为空则不回复
说明:
添加插件调用 次数 限制
参数:
:param plugin: 插件模块名称
:param max_count: 最大次数限制
:param status: 默认开关状态
:param limit_type: 限制类型 监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
:param rst: 回复的话，为空则不回复
"""
max_count = max_count or 5
status = status or True
limit_type = limit_type or "user"
if limit_type not in ["user", "group"]:
raise ValueError(f"{plugin} 添加count限制错误，‘limit_type‘ 必须为 'user'/'group'")
self._data[plugin] = {
"max_count": max_count,
"status": status,
"limit_type": limit_type,
"rst": rst,
}
self._data[plugin] = PluginCount(max_count=max_count, status=status, limit_type=limit_type, rst=rst)
def get_plugin_count_data(self, plugin: str) -> Optional[dict]:
def get_plugin_count_data(self, plugin: str) -> Optional[PluginCount]:
"""
获取插件次数数据
:param plugin: 模块名
说明:
获取插件次数数据
参数:
:param plugin: 模块名
"""
if self.check_plugin_count_status(plugin):
return self._data[plugin]
return None
def get_plugin_data(self, plugin: str) -> Optional[dict]:
def get_plugin_data(self, plugin: str) -> Optional[PluginCount]:
"""
获取单个模块限制数据
:param plugin: 模块名
说明:
获取单个模块限制数据
参数:
:param plugin: 模块名
"""
if self._data.get(plugin) is not None:
return self._data.get(plugin)
return None
def check_plugin_count_status(self, plugin: str) -> bool:
"""
检测插件是否有 次数 限制
:param plugin: 模块名
说明:
检测插件是否有 次数 限制
参数:
:param plugin: 模块名
"""
return (
plugin in self._data.keys()
and self._data[plugin]["status"]
and self._data[plugin]["max_count"] > 0
and self._data[plugin].status
and self._data[plugin].max_count > 0
)
def check(self, plugin: str, id_: int) -> bool:
"""
检查 count
:param plugin: 模块名
:param id_: 限制 id
说明:
检查 count
参数:
:param plugin: 模块名
:param id_: 限制 id
"""
if self._daily_limiter.get(plugin):
return self._daily_limiter[plugin].check(id_)
@@ -96,24 +90,25 @@ class Plugins2countManager(StaticData):
def increase(self, plugin: str, id_: int, num: int = 1):
"""
增加次数
:param plugin: 模块名
:param id_: cd 限制类型
:param num: 增加次数
:return:
说明:
增加次数
参数:
:param plugin: 模块名
:param id_: cd 限制类型
:param num: 增加次数
"""
if self._daily_limiter.get(plugin):
self._daily_limiter[plugin].increase(id_, num)
def reload_count_limit(self):
"""
加载 cd 限制器
:return:
说明:
加载 cd 限制器
"""
for plugin in self._data:
if self.check_plugin_count_status(plugin):
self._daily_limiter[plugin] = DailyNumberLimiter(
self.get_plugin_count_data(plugin)["max_count"]
self.get_plugin_count_data(plugin).max_count
)
logger.info(f"已成功加载 {len(self._daily_limiter)} 个Count限制.")
@@ -121,8 +116,59 @@ class Plugins2countManager(StaticData):
"""
重载本地数据
"""
self.__load_file()
self.reload_count_limit()
def save(self, path: Union[str, Path] = None):
"""
说明:
保存文件
参数:
:param path: 文件路径
"""
path = path or self.file
if isinstance(path, str):
path = Path(path)
if path:
with open(path, "w", encoding="utf8") as f:
yaml.dump(
{"PluginCountLimit": self.dict()},
f,
indent=2,
Dumper=yaml.RoundTripDumper,
allow_unicode=True,
)
_data = yaml.round_trip_load(open(path, encoding="utf8"))
_data["PluginCountLimit"].yaml_set_start_comment(
"""# 命令每日次数限制
# 即 用户/群聊 每日可调用命令的次数 [数据内存存储，重启将会重置]
# 每日调用直到 00:00 刷新
# key：模块名称
# max_count: 每日调用上限
# status：此限制的开关状态
# limit_type：监听对象，以user_id或group_id作为键来限制，'user'：用户id，'group'：群id
# 示例：'user'：用户上限，'group'：群聊上限
# rst：回复的话，可以添加[at]，[uname]，[nickname]来对应艾特，用户群名称，昵称系统昵称
# rst 为 "" 或 None 时则不回复
# rst示例："[uname]你冲的太快了，[nickname]先生，请稍后再冲[at]"
# rst回复："老色批你冲的太快了，欧尼酱先生，请稍后再冲@老色批"
# 用户昵称↑ 昵称系统的昵称↑ 艾特用户↑""",
indent=2,
)
with open(path, "w", encoding="utf8") as wf:
yaml.round_trip_dump(
_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True
)
def __load_file(self):
"""
说明:
读取配置文件
"""
self._data: Dict[str, PluginCount] = {}
if self.file.exists():
with open(self.file, "r", encoding="utf8") as f:
self._data: dict = yaml.load(f)
self._data = self._data["PluginCountLimit"]
self.reload_count_limit()
temp = _yaml.load(f)
if "PluginCountLimit" in temp.keys():
for k, v in temp["PluginCountLimit"].items():
self._data[k] = PluginCount.parse_obj(v)
utils/manager/plugins2settings_manager.py
Executable → Regular
+83 -60
View File
@@ -1,11 +1,11 @@
from typing import List, Optional, Union, Tuple
from .data_class import StaticData
from typing import List, Optional, Union, Tuple, Dict
from utils.manager.data_class import StaticData
from pathlib import Path
from ruamel.yaml import YAML
from ruamel import yaml
from .models import PluginSetting
_yaml = YAML(typ="safe")
_yaml = yaml.YAML(typ="safe")
class Plugins2settingsManager(StaticData):
@@ -14,97 +14,120 @@ class Plugins2settingsManager(StaticData):
"""
def __init__(self, file: Path):
self.file = file
super().__init__(None)
if file.exists():
with open(file, "r", encoding="utf8") as f:
self._data = _yaml.load(f)
if self._data:
if "PluginSettings" in self._data.keys():
self._data = (
self._data["PluginSettings"] if self._data["PluginSettings"] else {}
)
for x in self._data.keys():
if self._data[x].get("cost_gold") is None:
self._data[x]["cost_gold"] = 0
super().__init__(file, False)
self.__load_file()
def add_plugin_settings(
self,
plugin: str,
cmd: Optional[List[str]] = None,
default_status: Optional[bool] = True,
level: Optional[int] = 5,
limit_superuser: Optional[bool] = False,
default_status: bool = True,
level: int = 5,
limit_superuser: bool = False,
plugin_type: Tuple[Union[str, int]] = ("normal",),
cost_gold: int = 0,
):
"""
添加一个插件设置
:param plugin: 插件模块名称
:param cmd: 命令 或 命令别名
:param default_status: 默认开关状态
:param level: 功能权限等级
:param limit_superuser: 功能状态是否限制超级用户
:param plugin_type: 插件类型
:param cost_gold: 需要消费的金币
说明:
添加一个插件设置
参数:
:param plugin: 插件模块名称
:param cmd: 命令 或 命令别名
:param default_status: 默认开关状态
:param level: 功能权限等级
:param limit_superuser: 功能状态是否限制超级用户
:param plugin_type: 插件类型
:param cost_gold: 需要消费的金币
"""
level = level or 5
cmd = cmd or []
cost_gold = cost_gold or 0
self._data[plugin] = {
"level": level if level is not None else 5,
"default_status": default_status if default_status is not None else True,
"limit_superuser": (
limit_superuser
if limit_superuser is not None
else False
),
"cmd": cmd,
"plugin_type": list(
plugin_type if plugin_type is not None else ("normal",)
),
"cost_gold": cost_gold,
}
self._data[plugin] = PluginSetting(
cmd=cmd,
level=level,
default_status=default_status,
limit_superuser=limit_superuser,
plugin_type=plugin_type,
cost_gold=cost_gold,
)
def get_plugin_data(self, module: str) -> dict:
def get_plugin_data(self, module: str) -> Optional[PluginSetting]:
"""
通过模块名获取数据
:param module: 模块名称
说明:
通过模块名获取数据
参数:
:param module: 模块名称
"""
if self._data.get(module) is not None:
return self._data.get(module)
return {}
return self._data.get(module)
def get_plugin_module(
self, cmd: str, is_all: bool = False
) -> Union[str, List[str]]:
"""
根据 cmd 获取功能 modules
:param cmd: 命令
:param is_all: 获取全部包含cmd的模块
说明:
根据 cmd 获取功能 modules
参数:
:param cmd: 命令
:param is_all: 获取全部包含cmd的模块
"""
keys = []
for key in self._data.keys():
if cmd in self._data[key]["cmd"]:
if cmd in self._data[key].cmd:
if is_all:
keys.append(key)
else:
return key
return keys
def reload(self):
"""
说明:
重载本地数据
"""
self.__load_file()
def save(self, path: Union[str, Path] = None):
"""
说明:
保存文件
参数:
:param path: 文件路径
"""
path = path or self.file
if isinstance(path, str):
path = Path(path)
if path:
with open(path, "w", encoding="utf8") as f:
yaml.dump({"PluginSettings": self._data}, f, indent=2, Dumper=yaml.RoundTripDumper, allow_unicode=True)
yaml.dump(
{"PluginSettings": self.dict()},
f,
indent=2,
Dumper=yaml.RoundTripDumper,
allow_unicode=True,
)
_data = yaml.round_trip_load(open(path, encoding="utf8"))
_data["PluginSettings"].yaml_set_start_comment(
"""# 模块与对应命令和对应群权限
# 用于生成帮助图片 和 开关功能
# key：模块名称
# level：需要的群等级
# default_status：加入群时功能的默认开关状态
# limit_superuser: 功能状态是否限制超级用户
# cmd: 关闭[cmd] 都会触发命令 关闭对应功能，cmd列表第一个词为统计的功能名称
# plugin_type: 帮助类别 示例：('原神相关',) 或 ('原神相关', 1)，1代表帮助命令列向排列，否则为横向排列""",
indent=2,
)
with open(path, "w", encoding="utf8") as wf:
yaml.round_trip_dump(
_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True
)
def reload(self):
def __load_file(self):
"""
重载本地数据
说明:
读取配置文件
"""
self._data: Dict[str, PluginSetting] = {}
if self.file.exists():
with open(self.file, "r", encoding="utf8") as f:
self._data: dict = _yaml.load(f)
self._data = self._data["PluginSettings"]
temp = _yaml.load(f)
if "PluginSettings" in temp.keys():
for k, v in temp["PluginSettings"].items():
self._data[k] = PluginSetting.parse_obj(v)
utils/manager/plugins_manager.py
Executable → Regular
+66 -57
View File
@@ -1,7 +1,33 @@
from typing import Optional
from typing import Optional, Dict, Callable, Union
from pathlib import Path
from .data_class import StaticData
from . import group_manager
from utils.manager.data_class import StaticData
from utils.manager import group_manager
from .models import Plugin
def init_plugin(func: Callable):
"""
说明:
初始化群数据
参数:
:param func: func
"""
def wrapper(*args, **kwargs):
self = args[0]
module = args[1]
if module not in self._data.keys():
self._data[module] = Plugin(
plugin_name=module,
status=True,
error=False,
block_type=None,
author=None,
version=None,
)
return func(*args, **kwargs)
return wrapper
class PluginsManager(StaticData):
@@ -10,9 +36,10 @@ class PluginsManager(StaticData):
"""
def __init__(self, file: Path):
self._data: Dict[str, Plugin]
super().__init__(file)
if not self._data:
self._data = {}
for k, v in self._data.items():
self._data[k] = Plugin.parse_obj(v)
def add_plugin_data(
self,
@@ -26,23 +53,25 @@ class PluginsManager(StaticData):
version: Optional[int] = None,
):
"""
添加插件数据
:param module: 模块名称
:param plugin_name: 插件名称
:param status: 插件开关状态
:param error: 加载状态
:param block_type: 限制类型
:param author: 作者
:param version: 版本
说明:
添加插件数据
参数:
:param module: 模块名称
:param plugin_name: 插件名称
:param status: 插件开关状态
:param error: 加载状态
:param block_type: 限制类型
:param author: 作者
:param version: 版本
"""
self._data[module] = {
"plugin_name": plugin_name,
"status": status,
"error": error,
"block_type": block_type,
"author": author,
"version": version,
}
self._data[module] = Plugin(
plugin_name=plugin_name,
status=status,
error=error,
block_type=block_type,
author=author,
version=version,
)
def block_plugin(
self, module: str, group_id: Optional[int] = None, block_type: str = "all"
@@ -67,9 +96,7 @@ class PluginsManager(StaticData):
"""
self._set_plugin_status(module, "unblock", group_id)
def get_plugin_status(
self, module: str, block_type: str = "all"
) -> bool:
def get_plugin_status(self, module: str, block_type: str = "all") -> bool:
"""
说明:
获取插件状态
@@ -78,13 +105,12 @@ class PluginsManager(StaticData):
:param block_type: 限制类型
"""
if module in self._data.keys():
if self._data[module]["block_type"] == "all" and block_type == "all":
if self._data[module].block_type == "all" and block_type == "all":
return False
else:
return not self._data[module]["block_type"] == block_type
return not self._data[module].block_type == block_type
return True
def get_plugin_block_type(self, module: str) -> str:
def get_plugin_block_type(self, module: str) -> Optional[str]:
"""
说明:
获取功能限制类型
@@ -92,18 +118,19 @@ class PluginsManager(StaticData):
:param module: 模块名称
"""
if module in self._data.keys():
return self._data[module]["block_type"]
return ""
return self._data[module].block_type
@init_plugin
def get_plugin_error_status(self, module: str) -> bool:
"""
插件是否成功加载
:param module: 模块名称
说明:
插件是否成功加载
参数:
:param module: 模块名称
"""
if module not in self._data.keys():
self.init_plugin(module)
return self._data[module]["error"]
return self._data[module].error
@init_plugin
def _set_plugin_status(
self,
module: str,
@@ -120,7 +147,6 @@ class PluginsManager(StaticData):
:param group_id: 群组
:param block_type: 限制类型
"""
group_id = str(group_id) if group_id else group_id
if module:
if group_id:
if status == "block":
@@ -128,28 +154,11 @@ class PluginsManager(StaticData):
else:
group_manager.unblock_plugin(f"{module}:super", int(group_id))
else:
if module not in self._data.keys():
self.init_plugin(module)
if status == "block":
self._data[module]["status"] = False
self._data[module]["block_type"] = block_type
self._data[module].status = False
self._data[module].block_type = block_type
else:
if module in self._data.keys():
self._data[module]["status"] = True
self._data[module]["block_type"] = None
self._data[module].status = True
self._data[module].block_type = None
self.save()
def init_plugin(self, module: str):
"""
初始化插件数据
:param module: 模块名称
"""
if module not in self._data.keys():
self._data[module] = {
"plugin_name": module,
"status": True,
"error": False,
"block_type": None,
"author": None,
"version": None,
}
utils/manager/requests_manager.py
Executable → Regular
+271 -271
View File
@@ -1,271 +1,271 @@
from utils.manager.data_class import StaticData
from nonebot.adapters.onebot.v11 import Bot, ActionFailed
from services.log import logger
from typing import Optional, Literal
from utils.image_utils import BuildImage
from utils.utils import get_user_avatar
from pathlib import Path
from io import BytesIO
class RequestManager(StaticData):
"""
好友请求/邀请请求 管理
"""
def __init__(self, file: Optional[Path]):
super().__init__(file)
if not self._data:
self._data = {"private": {}, "group": {}}
def add_request(
self,
id_: int,
type_: str,
flag: str,
*,
nickname: Optional[str] = None,
level: Optional[int] = None,
sex: Optional[str] = None,
age: Optional[str] = None,
from_: Optional[str] = "",
comment: Optional[str] = None,
invite_group: Optional[int] = None,
group_name: Optional[str] = None,
):
"""
添加一个请求
:param id_: id，用户id或群id
:param type_: 类型，private 或 group
:param flag: event.flag
:param nickname: 用户昵称
:param level: 等级
:param sex: 性别
:param age: 年龄
:param from_: 请求来自
:param comment: 附加消息
:param invite_group: 邀请群聊
:param group_name: 群聊名称
"""
self._data[type_][str(len(self._data[type_].keys()))] = {
"id": id_,
"flag": flag,
"nickname": nickname,
"level": level,
"sex": sex,
"age": age,
"from": from_,
"comment": comment,
"invite_group": invite_group,
"group_name": group_name,
}
self.save()
def remove_request(self, type_: str, id_: int):
"""
删除一个请求数据
:param type_: 类型
:param id_: id，user_id 或 group_id
"""
for x in self._data[type_].keys():
if self._data[type_][x].get("id") == id_:
del self._data[type_][x]
break
self.save()
def get_group_id(self, id_: int) -> Optional[int]:
"""
通过id获取群号
:param id_: id
"""
data = self._data["group"].get(str(id_))
if data:
return data["invite_group"]
return None
async def approve(self, bot: Bot, id_: int, type_: str) -> int:
"""
同意请求
:param bot: Bot
:param id_: id
:param type_: 类型，private 或 group
"""
return await self._set_add_request(bot, id_, type_, True)
async def refused(self, bot: Bot, id_: int, type_: str) -> Optional[int]:
"""
拒绝请求
:param bot: Bot
:param id_: id
:param type_: 类型，private 或 group
"""
return await self._set_add_request(bot, id_, type_, False)
def clear(self, type_: Optional[str] = None): # type_: Optional[Literal["group", "private"]] = None
"""
清空所有请求信息，无视请求
:param type_: 类型
"""
if type_:
self._data[type_] = {}
else:
self._data = {"private": {}, "group": {}}
self.save()
def delete_request(self, id_: int, type_: str): # type_: Literal["group", "private"]
"""
删除请求
:param id_: id
:param type_: 类型
"""
id_ = str(id_)
if self._data[type_].get(id_):
del self._data[type_][id_]
self.save()
def set_group_name(self, group_name: str, group_id: int):
"""
设置群聊名称
:param group_name: 名称
:param group_id: id
"""
for id_ in self._data["group"].keys():
if self._data["group"][id_]["invite_group"] == group_id:
self._data["group"][id_]["group_name"] = group_name
break
self.save()
async def show(self, type_: str) -> Optional[str]:
"""
请求可视化
"""
data = self._data[type_]
if not data:
return None
img_list = []
id_list = list(data.keys())
id_list.reverse()
for id_ in id_list:
age = data[id_]["age"]
nickname = data[id_]["nickname"]
comment = data[id_]["comment"] if type_ == "private" else ""
from_ = data[id_]["from"]
sex = data[id_]["sex"]
ava = BuildImage(
80, 80, background=BytesIO(await get_user_avatar(data[id_]["id"]))
)
ava.circle()
age_bk = BuildImage(
len(str(age)) * 6 + 6,
15,
color="#04CAF7" if sex == "male" else "#F983C1",
)
age_bk.text((3, 1), f"{age}", fill=(255, 255, 255))
x = BuildImage(
90, 32, font_size=15, color="#EEEFF4", font="HYWenHei-85W.ttf"
)
x.text((0, 0), "同意/拒绝", center_type="center")
x.circle_corner(10)
A = BuildImage(500, 100, font_size=24, font="msyh.ttf")
A.paste(ava, (15, 0), alpha=True, center_type="by_height")
A.text((120, 15), nickname)
A.paste(age_bk, (120, 50), True)
A.paste(
BuildImage(
200,
0,
font_size=12,
plain_text=f"对方留言：{comment}",
font_color=(140, 140, 143),
),
(120 + age_bk.w + 10, 49),
True,
)
if type_ == "private":
A.paste(
BuildImage(
200,
0,
font_size=12,
plain_text=f"来源：{from_}",
font_color=(140, 140, 143),
),
(120, 70),
True,
)
else:
A.paste(
BuildImage(
200,
0,
font_size=12,
plain_text=f"邀请你加入：{data[id_]['group_name']}({data[id_]['invite_group']})",
font_color=(140, 140, 143),
),
(120, 70),
True,
)
A.paste(x, (380, 35), True)
A.paste(
BuildImage(
0,
0,
plain_text=f"id：{id_}",
font_size=13,
font_color=(140, 140, 143),
),
(400, 10),
True,
)
img_list.append(A)
A = BuildImage(500, len(img_list) * 100, 500, 100)
for img in img_list:
A.paste(img)
bk = BuildImage(A.w, A.h + 50, color="#F8F9FB", font_size=20)
bk.paste(A, (0, 50))
bk.text(
(15, 13), "好友请求" if type_ == "private" else "群聊请求", fill=(140, 140, 143)
)
return bk.pic2bs4()
async def _set_add_request(
self, bot: Bot, id_: int, type_: str, approve: bool
) -> int:
"""
处理请求
:param bot: Bot
:param id_: id
:param type_: 类型，private 或 group
:param approve: 是否同意
"""
id_ = str(id_)
if id_ in self._data[type_].keys():
try:
if type_ == "private":
await bot.set_friend_add_request(
flag=self._data[type_][id_]["flag"], approve=approve
)
rid = self._data[type_][id_]["id"]
else:
await bot.set_group_add_request(
flag=self._data[type_][id_]["flag"],
sub_type="invite",
approve=approve,
)
rid = self._data[type_][id_]["invite_group"]
except ActionFailed:
logger.info(
f"同意{self._data[type_][id_]['nickname']}({self._data[type_][id_]['id']})"
f"的{'好友' if type_ == 'private' else '入群'}请求失败了..."
)
return 1 # flag失效
else:
logger.info(
f"{'同意' if approve else '拒绝'}{self._data[type_][id_]['nickname']}({self._data[type_][id_]['id']})"
f"的{'好友' if type_ == 'private' else '入群'}请求..."
)
del self._data[type_][id_]
self.save()
return rid
return 2 # 未找到id
from utils.manager.data_class import StaticData
from nonebot.adapters.onebot.v11 import Bot, ActionFailed
from services.log import logger
from typing import Optional
from utils.image_utils import BuildImage
from utils.utils import get_user_avatar
from pathlib import Path
from io import BytesIO
class RequestManager(StaticData):
"""
好友请求/邀请请求 管理
"""
def __init__(self, file: Optional[Path]):
super().__init__(file)
if not self._data:
self._data = {"private": {}, "group": {}}
def add_request(
self,
id_: int,
type_: str,
flag: str,
*,
nickname: Optional[str] = None,
level: Optional[int] = None,
sex: Optional[str] = None,
age: Optional[str] = None,
from_: Optional[str] = "",
comment: Optional[str] = None,
invite_group: Optional[int] = None,
group_name: Optional[str] = None,
):
"""
添加一个请求
:param id_: id，用户id或群id
:param type_: 类型，private 或 group
:param flag: event.flag
:param nickname: 用户昵称
:param level: 等级
:param sex: 性别
:param age: 年龄
:param from_: 请求来自
:param comment: 附加消息
:param invite_group: 邀请群聊
:param group_name: 群聊名称
"""
self._data[type_][str(len(self._data[type_].keys()))] = {
"id": id_,
"flag": flag,
"nickname": nickname,
"level": level,
"sex": sex,
"age": age,
"from": from_,
"comment": comment,
"invite_group": invite_group,
"group_name": group_name,
}
self.save()
def remove_request(self, type_: str, id_: int):
"""
删除一个请求数据
:param type_: 类型
:param id_: id，user_id 或 group_id
"""
for x in self._data[type_].keys():
if self._data[type_][x].get("id") == id_:
del self._data[type_][x]
break
self.save()
def get_group_id(self, id_: int) -> Optional[int]:
"""
通过id获取群号
:param id_: id
"""
data = self._data["group"].get(str(id_))
if data:
return data["invite_group"]
return None
async def approve(self, bot: Bot, id_: int, type_: str) -> int:
"""
同意请求
:param bot: Bot
:param id_: id
:param type_: 类型，private 或 group
"""
return await self._set_add_request(bot, id_, type_, True)
async def refused(self, bot: Bot, id_: int, type_: str) -> Optional[int]:
"""
拒绝请求
:param bot: Bot
:param id_: id
:param type_: 类型，private 或 group
"""
return await self._set_add_request(bot, id_, type_, False)
def clear(self, type_: Optional[str] = None): # type_: Optional[Literal["group", "private"]] = None
"""
清空所有请求信息，无视请求
:param type_: 类型
"""
if type_:
self._data[type_] = {}
else:
self._data = {"private": {}, "group": {}}
self.save()
def delete_request(self, id_: int, type_: str): # type_: Literal["group", "private"]
"""
删除请求
:param id_: id
:param type_: 类型
"""
id_ = str(id_)
if self._data[type_].get(id_):
del self._data[type_][id_]
self.save()
def set_group_name(self, group_name: str, group_id: int):
"""
设置群聊名称
:param group_name: 名称
:param group_id: id
"""
for id_ in self._data["group"].keys():
if self._data["group"][id_]["invite_group"] == group_id:
self._data["group"][id_]["group_name"] = group_name
break
self.save()
async def show(self, type_: str) -> Optional[str]:
"""
请求可视化
"""
data = self._data[type_]
if not data:
return None
img_list = []
id_list = list(data.keys())
id_list.reverse()
for id_ in id_list:
age = data[id_]["age"]
nickname = data[id_]["nickname"]
comment = data[id_]["comment"] if type_ == "private" else ""
from_ = data[id_]["from"]
sex = data[id_]["sex"]
ava = BuildImage(
80, 80, background=BytesIO(await get_user_avatar(data[id_]["id"]))
)
ava.circle()
age_bk = BuildImage(
len(str(age)) * 6 + 6,
15,
color="#04CAF7" if sex == "male" else "#F983C1",
)
age_bk.text((3, 1), f"{age}", fill=(255, 255, 255))
x = BuildImage(
90, 32, font_size=15, color="#EEEFF4", font="HYWenHei-85W.ttf"
)
x.text((0, 0), "同意/拒绝", center_type="center")
x.circle_corner(10)
A = BuildImage(500, 100, font_size=24, font="msyh.ttf")
A.paste(ava, (15, 0), alpha=True, center_type="by_height")
A.text((120, 15), nickname)
A.paste(age_bk, (120, 50), True)
A.paste(
BuildImage(
200,
0,
font_size=12,
plain_text=f"对方留言：{comment}",
font_color=(140, 140, 143),
),
(120 + age_bk.w + 10, 49),
True,
)
if type_ == "private":
A.paste(
BuildImage(
200,
0,
font_size=12,
plain_text=f"来源：{from_}",
font_color=(140, 140, 143),
),
(120, 70),
True,
)
else:
A.paste(
BuildImage(
200,
0,
font_size=12,
plain_text=f"邀请你加入：{data[id_]['group_name']}({data[id_]['invite_group']})",
font_color=(140, 140, 143),
),
(120, 70),
True,
)
A.paste(x, (380, 35), True)
A.paste(
BuildImage(
0,
0,
plain_text=f"id：{id_}",
font_size=13,
font_color=(140, 140, 143),
),
(400, 10),
True,
)
img_list.append(A)
A = BuildImage(500, len(img_list) * 100, 500, 100)
for img in img_list:
A.paste(img)
bk = BuildImage(A.w, A.h + 50, color="#F8F9FB", font_size=20)
bk.paste(A, (0, 50))
bk.text(
(15, 13), "好友请求" if type_ == "private" else "群聊请求", fill=(140, 140, 143)
)
return bk.pic2bs4()
async def _set_add_request(
self, bot: Bot, id_: int, type_: str, approve: bool
) -> int:
"""
处理请求
:param bot: Bot
:param id_: id
:param type_: 类型，private 或 group
:param approve: 是否同意
"""
id_ = str(id_)
if id_ in self._data[type_].keys():
try:
if type_ == "private":
await bot.set_friend_add_request(
flag=self._data[type_][id_]["flag"], approve=approve
)
rid = self._data[type_][id_]["id"]
else:
await bot.set_group_add_request(
flag=self._data[type_][id_]["flag"],
sub_type="invite",
approve=approve,
)
rid = self._data[type_][id_]["invite_group"]
except ActionFailed:
logger.info(
f"同意{self._data[type_][id_]['nickname']}({self._data[type_][id_]['id']})"
f"的{'好友' if type_ == 'private' else '入群'}请求失败了..."
)
return 1 # flag失效
else:
logger.info(
f"{'同意' if approve else '拒绝'}{self._data[type_][id_]['nickname']}({self._data[type_][id_]['id']})"
f"的{'好友' if type_ == 'private' else '入群'}请求..."
)
del self._data[type_][id_]
self.save()
return rid
return 2 # 未找到id
utils/manager/resources_manager.py
Executable → Regular
+1 -1
View File
@@ -1,7 +1,7 @@
from typing import Union, List, Optional
from configs.path_config import IMAGE_PATH, DATA_PATH, RECORD_PATH, TEXT_PATH, FONT_PATH, LOG_PATH
from .data_class import StaticData
from utils.manager.data_class import StaticData
from pathlib import Path
from ruamel.yaml import YAML
from services.log import logger
utils/manager/withdraw_message_manager.py
Executable → Regular
View File