Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2025-12-15 14:22:55 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

修复web_ui群组无法获取，修复web_ui修改插件数据时cmd格式错误

Browse Source
This commit is contained in:
HibiKier 2022-11-30 23:40:55 +08:00
parent 7b3a3451f6
commit 994484f3b0
14 changed files with 109 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
README.md
View File

@ -296,6 +296,15 @@ PS: **ARM平台** 请使用全量版 同时 **如果你的机器 RAM < 1G 可能
## 更新
### 2022/11/28
* 修复web_ui群组无法获取
* 修复web_ui修改插件数据时cmd格式错误
### 2022/11/28
* :bug: Fix a bug in open_cases to get vanilla knives' prices [@pull/1188](https://github.com/HibiKier/zhenxun_bot/pull/1188)
### 2022/11/24
* 修复管理员插件加载路径错误

112
basic_plugins/admin_bot_manage/_data_source.py
View File

@ -16,53 +16,55 @@ from utils.http_utils import AsyncHttpx
import asyncio
import time
import os
import ujson as json
try:
import ujson as json
except ModuleNotFoundError:
import json
custom_welcome_msg_json = (
Path() / "data" / "custom_welcome_msg" / "custom_welcome_msg.json"
)
def group_current_status(group_id: int) -> str:
"""
获取当前所有通知的开关
:param group_id: 群号
说明:
获取当前群聊所有通知的开关
参数:
:param group_id: 群号
"""
rst = "[被动技能 状态]\n"
_data = group_manager.get_task_data()
for task in _data.keys():
rst += f'{_data[task]}: {"√" if group_manager.check_group_task_status(group_id, task) else "×"}\n'
return rst.strip()
custom_welcome_msg_json = (
Path() / "data" / "custom_welcome_msg" / "custom_welcome_msg.json"
)
return "[被动技能 状态]\n" + "\n".join(
[
f'{_data[task]}: {"√" if group_manager.check_group_task_status(group_id, task) else "×"}\n'
for task in _data
]
)
async def custom_group_welcome(
msg: str, imgs: List[str], user_id: int, group_id: int
msg: str, img_list: List[str], user_id: int, group_id: int
) -> str:
"""
替换群欢迎消息
:param msg: 欢迎消息文本
:param imgs: 欢迎消息图片，只取第一张
:param user_id: 用户id，用于log记录
:param group_id: 群号
说明:
替换群欢迎消息
参数:
:param msg: 欢迎消息文本
:param img_list: 欢迎消息图片，只取第一张
:param user_id: 用户id，用于log记录
:param group_id: 群号
"""
img_result = ""
img = imgs[0] if imgs else ""
result = ""
img = img_list[0] if img_list else ""
if (DATA_PATH / f"custom_welcome_msg/{group_id}.jpg").exists():
(DATA_PATH / f"custom_welcome_msg/{group_id}.jpg").unlink()
data = {}
if not custom_welcome_msg_json.exists():
custom_welcome_msg_json.parent.mkdir(parents=True, exist_ok=True)
data = {}
else:
try:
data = json.load(open(custom_welcome_msg_json, "r"))
except FileNotFoundError:
data = {}
pass
try:
if msg:
data[str(group_id)] = str(msg)
@ -88,10 +90,12 @@ task_data = None
async def change_group_switch(cmd: str, group_id: int, is_super: bool = False):
"""
修改群功能状态
:param cmd: 功能名称
:param group_id: 群号
:param is_super: 是否位超级用户，超级用户用于私聊开关功能状态
说明:
修改群功能状态
参数:
:param cmd: 功能名称
:param group_id: 群号
:param is_super: 是否位超级用户，超级用户用于私聊开关功能状态
"""
global task_data
if not task_data:
@ -161,9 +165,11 @@ async def change_group_switch(cmd: str, group_id: int, is_super: bool = False):
def set_plugin_status(cmd: str, block_type: str = "all"):
"""
设置插件功能状态（超级用户使用）
:param cmd: 功能名称
:param block_type: 限制类型, 'all': 私聊+群里, 'private': 私聊, 'group': 群聊
说明:
设置插件功能状态（超级用户使用）
参数:
:param cmd: 功能名称
:param block_type: 限制类型, 'all': 私聊+群里, 'private': 私聊, 'group': 群聊
"""
status = cmd[:2]
cmd = cmd[2:]
@ -179,14 +185,16 @@ def set_plugin_status(cmd: str, block_type: str = "all"):
async def get_plugin_status():
"""
获取功能状态
说明:
获取功能状态
"""
return await asyncio.get_event_loop().run_in_executor(None, _get_plugin_status)
def _get_plugin_status() -> MessageSegment:
"""
合成功能状态图片
说明:
合成功能状态图片
"""
rst = "\t功能\n"
flag_str = "状态".rjust(4) + "\n"
@ -200,9 +208,9 @@ def _get_plugin_status() -> MessageSegment:
try:
plugin_name = plugins_manager.get(module).plugin_name
if (
"[Hidden]" in plugin_name
or "[Admin]" in plugin_name
or "[Superuser]" in plugin_name
"[Hidden]" in plugin_name
or "[Admin]" in plugin_name
or "[Superuser]" in plugin_name
):
continue
rst += f"{plugin_name}"
@ -225,9 +233,11 @@ def _get_plugin_status() -> MessageSegment:
async def update_member_info(group_id: int, remind_superuser: bool = False) -> bool:
"""
更新群成员信息
:param group_id: 群号
:param remind_superuser: 失败信息提醒超级用户
说明:
更新群成员信息
参数:
:param group_id: 群号
:param remind_superuser: 失败信息提醒超级用户
"""
bot = get_bot()
_group_user_list = await bot.get_group_member_list(group_id=group_id)
@ -238,14 +248,10 @@ async def update_member_info(group_id: int, remind_superuser: bool = False) -> b
nickname = user_info["card"] or user_info["nickname"]
async with db.transaction():
# 更新权限
if (
user_info["role"]
in [
if user_info["role"] in [
"owner",
"admin",
]
and not await LevelUser.is_group_flag(user_info["user_id"], group_id)
):
] and not await LevelUser.is_group_flag(user_info["user_id"], group_id):
await LevelUser.set_level(
user_info["user_id"],
user_info["group_id"],
@ -273,10 +279,10 @@ async def update_member_info(group_id: int, remind_superuser: bool = False) -> b
"%Y-%m-%d %H:%M:%S",
)
if await GroupInfoUser.add_member_info(
user_info["user_id"],
user_info["group_id"],
nickname,
join_time,
user_info["user_id"],
user_info["group_id"],
nickname,
join_time,
):
_exist_member_list.append(int(user_info["user_id"]))
logger.info(f"用户{user_info['user_id']} 所属{user_info['group_id']} 更新成功")
@ -307,9 +313,11 @@ async def update_member_info(group_id: int, remind_superuser: bool = False) -> b
def set_group_bot_status(group_id: int, status: bool) -> str:
"""
设置群聊bot开关状态
:param group_id: 群号
:param status: 状态
说明:
设置群聊bot开关状态
参数:
:param group_id: 群号
:param status: 状态
"""
if status:
if group_manager.check_group_bot_status(group_id):

7
basic_plugins/admin_bot_manage/custom_welcome_message.py
View File

@ -1,11 +1,13 @@
from typing import List
from nonebot import on_command
from utils.utils import get_message_img
from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message
from nonebot.params import CommandArg
from ._data_source import custom_group_welcome
from nonebot.adapters.onebot.v11.permission import GROUP
from configs.config import Config
from services.log import logger
from utils.depends import ImageList
__zx_plugin_name__ = "自定义进群欢迎消息 [Admin]"
@ -35,9 +37,8 @@ custom_welcome = on_command(
@custom_welcome.handle()
async def _(event: GroupMessageEvent, arg: Message = CommandArg()):
async def _(event: GroupMessageEvent, arg: Message = CommandArg(), img: List[str] = ImageList):
msg = arg.extract_plain_text().strip()
img = get_message_img(event.json())
if not msg and not img:
await custom_welcome.finish(__plugin_usage__)
try:

6
basic_plugins/ban/__init__.py
View File

@ -1,4 +1,4 @@
from typing import Tuple
from typing import Tuple, List
from configs.config import NICKNAME, Config
from models.ban_user import BanUser
@ -9,6 +9,7 @@ from nonebot.adapters.onebot.v11 import (Bot, GroupMessageEvent, Message,
from nonebot.params import Command, CommandArg
from nonebot.permission import SUPERUSER
from services.log import logger
from utils.depends import AtList
from utils.utils import get_message_at, is_number
from .data_source import a_ban, parse_ban_time
@ -69,10 +70,10 @@ async def _(
event: GroupMessageEvent,
cmd: Tuple[str, ...] = Command(),
arg: Message = CommandArg(),
qq: List[str] = AtList()
):
cmd = cmd[0]
result = ""
qq = get_message_at(event.json())
if qq:
qq = qq[0]
user_name = await bot.get_group_member_info(group_id=event.group_id, user_id=qq)
@ -127,7 +128,6 @@ async def _(
if str(event.user_id) in bot.config.superusers:
msg_splt = msg.split()
if is_number(msg_splt[0]):
qq = int(msg_splt[0])
msg = msg_splt[1:]
if cmd in [".ban", "/ban"]:

3
basic_plugins/broadcast/__init__.py
View File

@ -40,10 +40,9 @@ async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
rst = ""
for img in img_list:
rst += image(img)
gl = await bot.get_group_list()
gl = [
g["group_id"]
for g in gl
for g in await bot.get_group_list()
if group_manager.check_group_task_status(g["group_id"], "broadcast")
]
g_cnt = len(gl)

3
basic_plugins/chat_history/__init__.py
View File

@ -1,3 +1,4 @@
import nonebot
from pathlib import Path
nonebot.load_plugins("basic_plugins/chat_history")
nonebot.load_plugins(str(Path(__file__).parent.resolve()))

8
basic_plugins/chat_history/chat_message.py
View File

@ -2,7 +2,7 @@ from configs.config import Config
from models.chat_history import ChatHistory
from nonebot import on_message
from nonebot.adapters.onebot.v11 import GroupMessageEvent, MessageEvent
from utils.utils import get_message_text
from utils.depends import PlaintText
from ._rule import rule
@ -20,13 +20,13 @@ chat_history = on_message(rule=rule, priority=1, block=False)
@chat_history.handle()
async def _(event: MessageEvent):
async def _(event: MessageEvent, msg: str = PlaintText()):
if isinstance(event, GroupMessageEvent):
await ChatHistory.add_chat_msg(
event.user_id, event.group_id, str(event.get_message()), get_message_text(event.json())
event.user_id, event.group_id, str(event.get_message()), msg
)
else:
await ChatHistory.add_chat_msg(event.user_id, None, str(event.get_message()), get_message_text(event.json()))
await ChatHistory.add_chat_msg(event.user_id, None, str(event.get_message()), msg)
# @test.handle()

6
basic_plugins/group_handle/__init__.py
View File

@ -19,11 +19,7 @@ from configs.config import Config
from pathlib import Path
import random
import os
try:
import ujson as json
except ModuleNotFoundError:
import json
import ujson as json
__zx_plugin_name__ = "群事件处理 [Hidden]"

11
basic_plugins/hooks/ban_hook.py
View File

@ -29,10 +29,13 @@ _flmt = FreqLimiter(300)
# 检查是否被ban
@run_preprocessor
async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
if (
(isinstance(event, MessageEvent) or isinstance(event, PokeNotifyEvent))
and matcher.priority not in [1, 999]
) or matcher.plugin_name in other_limit_plugins:
if hasattr(event, "user_id") and (
(
(isinstance(event, MessageEvent) or isinstance(event, PokeNotifyEvent))
and matcher.priority not in [1, 999]
)
or matcher.plugin_name in other_limit_plugins
):
try:
if (
await BanUser.is_super_ban(event.user_id)

3
basic_plugins/super_cmd/__init__.py
View File

@ -1,7 +1,8 @@
import nonebot
from pathlib import Path
nonebot.load_plugins('basic_plugins/super_cmd')
nonebot.load_plugins(str(Path(__file__).parent.resolve()))

1
plugins/ai/__init__.py
View File

@ -1,3 +1,4 @@
from nonebot import on_message
from nonebot.adapters.onebot.v11 import (
Bot,

4
plugins/web_ui/api/group.py
View File

@ -19,10 +19,10 @@ async def _(user: User = Depends(token_to_user)) -> Result:
for g in group_list:
group_info[g["group_id"]] = Group(**g)
group_data = group_manager.get_data()
for group_id in group_data["group_manager"]:
for group_id in group_data.group_manager:
try:
task_list = []
data = group_data["group_manager"][group_id]
data = group_manager[group_id].dict()
for tn, status in data["group_task_status"].items():
task_list.append(
Task(

3
plugins/web_ui/api/plugins.py
View File

@ -159,6 +159,9 @@ def _(plugin: Plugin, user: User = Depends(token_to_user)) -> Result:
code=500,
data=f"WEB_UI POST /webui/plugins model：{plugin.model} 发生错误 {type(e)}：{e}",
)
for key in plugins2settings_manager.keys():
if isinstance(plugins2settings_manager[key].cmd, str):
plugins2settings_manager[key].cmd = plugins2settings_manager[key].cmd.split(',')
plugins2settings_manager.save()
plugins_manager.save()
return Result(code=200, data="修改成功！")

10
utils/manager/group_manager.py
View File

@ -1,3 +1,4 @@
import copy
from typing import List, Union, Dict, Callable
from pathlib import Path
from .models import BaseData, BaseGroup
@ -65,6 +66,9 @@ class GroupManager(StaticData):
super().__init__(file, False)
self._data = BaseData.parse_file(file) if file.exists() else BaseData()
def get_data(self) -> BaseData:
return copy.deepcopy(self._data)
def block_plugin(self, module: str, group_id: int, is_save: bool = True):
"""
说明:
@ -340,6 +344,12 @@ class GroupManager(StaticData):
with open(path, "w", encoding="utf8") as f:
json.dump(dict_data, f, ensure_ascii=False, indent=4)
def __setitem__(self, key, value):
self._data.group_manager[key] = value
def __getitem__(self, key) -> BaseGroup:
return self._data.group_manager[key]
# def get_super_old_data(self) -> Optional[dict]:
# """
# 说明: