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

Merge pull request #160 from yajiwa/yajiwa

Browse Source 
update word_bank
This commit is contained in:
HibiKier
2022-05-16 19:19:20 +08:00
committed by GitHub
parent 4b3e28e14b 71ce9153ed
commit e27809298e
6 changed files with 177 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/word_bank/_data_source.py
+11 -5
View File
@@ -7,9 +7,8 @@ class WordBankBuilder:
def __init__(self, user_id: int, group_id: int, problem: str):
self._data = {
"user_id": user_id,
"group_id": group_id,
"problem": problem
}
"group_id": group_id}
self.problem = problem
def set_placeholder(self, id_: int, placeholder: Union[str, int]):
"""
@@ -28,13 +27,20 @@ class WordBankBuilder:
"""
self._data["answer"] = answer
async def save(self,search_type):
def set_problem(self, problem: str):
"""
设置问题
:param problem: 问题
"""
self._data["problem"] = problem
async def save(self, search_type):
user_id = self._data["user_id"]
group_id = self._data["group_id"]
problem = self._data["problem"]
answer = self._data["answer"]
placeholder = self._data.get("placeholder")
return await WordBank.add_problem_answer(user_id, group_id, search_type,problem, answer, placeholder)
return await WordBank.add_problem_answer(user_id, group_id, search_type, problem, answer, placeholder)
async def update(self, index):
user_id = self._data["user_id"]
plugins/word_bank/_rule.py
+11 -2
View File
@@ -1,12 +1,21 @@
import re
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, Event
from utils.utils import get_message_text
from utils.utils import get_message_img_file
from nonebot.typing import T_State
from .model import WordBank
async def check(bot: Bot, event: Event, state: T_State) -> bool:
if isinstance(event, GroupMessageEvent):
msg = event.raw_message
list_img = get_message_img_file(event.json())
if list_img:
for img_file in list_img:
strinfo = re.compile(f"{img_file},subType=\d*]")
msg = strinfo.sub(f'{img_file}]', msg)
print(msg)
return bool(
await WordBank.check(event.group_id, get_message_text(event.json()), event.is_tome())
await WordBank.check(event.group_id, msg, event.is_tome())
)
return False
plugins/word_bank/message_handle.py
+75 -2
View File
@@ -3,9 +3,11 @@ from ._rule import check
from .model import WordBank
from configs.path_config import DATA_PATH
from nonebot.adapters.onebot.v11 import GroupMessageEvent
from utils.utils import get_message_text
from utils.utils import get_message_at, get_message_img
from nonebot import on_message
from models.group_member_info import GroupInfoUser
from utils.utils import get_message_img_file, is_number
import re
__zx_plugin_name__ = "词库问答回复操作 [Hidden]"
@@ -17,8 +19,14 @@ message_handle = on_message(priority=6, block=True, rule=check)
@message_handle.handle()
async def _(event: GroupMessageEvent):
msg = event.raw_message
list_img = get_message_img_file(event.json())
if list_img:
for img_file in list_img:
strinfo = re.compile(f"{img_file},subType=\d*]")
msg = strinfo.sub(f'{img_file}]', msg)
q = await WordBank.check(
event.group_id, get_message_text(event.json()), event.is_tome()
event.group_id, msg, event.is_tome()
)
await message_handle.send(await get_one_answer(event, q.format, q.answer))
@@ -54,3 +62,68 @@ async def get_one_answer(event, format, _answer, all=1):
answer += _a[: _a.find(f"[__placeholder_{idx}]")] + "@" + q.user_name
_a = _a[_a.find(f"[__placeholder_{idx}]") + len(f"[__placeholder_{idx}]"):]
return answer + _a
# 处理单条问题
async def get_one_problem(event, problem):
strinfo = re.compile(f",subType=\d")
problem = strinfo.sub('', problem)
_problem = problem
_p = problem
problem = ''
for img in get_message_img(event.json()):
_x = img.split("?")[0]
r = re.search(rf"\[CQ:image,file=(.*),url={_x}.*?]", _p)
if r:
_problem = _problem.replace(
rf",url={img}",
f"",
)
problem += _p[: _p.find(f"[CQ:image,file={r.group(1)},url={img}]")] + image(img)
_p = _p[
_p.find(f"[CQ:image,file={r.group(1)},url={img}]") + len(f"[CQ:image,file={r.group(1)},url={img}]"):]
for at_ in get_message_at(event.json()):
r = re.search(rf"\[CQ:at,qq={at_}]", problem)
if r:
q = await GroupInfoUser.get_member_info(
int(at_), event.group_id)
problem += _p[: _p.find(f"[CQ:at,qq={at_}]")] + "@" + q.user_name
_p = _p[_p.find(f"[CQ:at,qq={at_}]") + len(f"[CQ:at,qq={at_}]"):]
return _problem, problem + _p
# 显示单条数据库问题
async def get_one_image_problem(event, problem):
path = data_dir / f"{event.group_id}" / "problem"
placeholder_list = []
idx = 0
img_list = re.findall(rf"\[CQ:image,file=(.*?)]", problem)
at_list = re.findall(rf"\[CQ:at,qq=(.*?)]", problem)
if img_list:
for img in img_list:
problem = problem.replace(f'[CQ:image,file={img}]', f'[__placeholder_{idx}]', 1)
placeholder_list.append([idx, img])
idx += 1
if at_list:
for ats in at_list:
problem = problem.replace(f'[CQ:at,qq={ats}]', f'[__placeholder_{idx}]', 1)
placeholder_list.append([idx, ats])
idx += 1
_p = problem
problem = ''
if not placeholder_list:
problem = _p
return problem
else:
for idx, placeholder in placeholder_list:
if is_number(placeholder):
q = await GroupInfoUser.get_member_info(
int(placeholder), event.group_id)
problem += _p[: _p.find(f"[__placeholder_{idx}]")] + "@" + q.user_name
else:
problem += _p[: _p.find(f"[__placeholder_{idx}]")] + image(
path / f"{placeholder}.jpg"
)
_p = _p[_p.find(f"[__placeholder_{idx}]") + len(f"[__placeholder_{idx}]"):]
return problem + _p
plugins/word_bank/model.py
-5
View File
@@ -108,11 +108,6 @@ class WordBank(db.Model):
:param group_id: 群号
:param problem: 问题
"""
if problem.startswith("id:"):
problem_index = int(problem.split(":")[-1])
q = await cls.get_group_all_problem(group_id)
if len(q) > problem_index:
problem = q[problem_index]
q = await cls.query.where(
(cls.group_id == group_id) & (cls.problem == problem)
).gino.all()
plugins/word_bank/word_hanlde.py
+55 -22
View File
@@ -6,7 +6,7 @@ from utils.http_utils import AsyncHttpx
from ._data_source import WordBankBuilder
from utils.message_builder import image
from utils.image_utils import text2image
from .message_handle import get_one_answer
from .message_handle import get_one_answer, get_one_problem, get_one_image_problem
from .model import WordBank
from nonebot.adapters.onebot.v11 import (
Bot,
@@ -19,6 +19,7 @@ import random
import os
import re
from configs.config import NICKNAME, Config
from models.group_member_info import GroupInfoUser
__zx_plugin_name__ = "词库问答 [Admin]"
__plugin_usage__ = """
@@ -26,7 +27,7 @@ usage：
对指定问题的随机回答，对相同问题可以设置多个不同回答
删除词条后每个词条的id可能会变化，请查看后再删除
指令：
添加词条 ?[模糊/关键字]...答...：添加问答词条，可重复添加相同问题的不同回答
添加词条 ?[模糊/关键字|词]...答...：添加问答词条，可重复添加相同问题的不同回答
删除词条 [问题/下标] ?[下标]：删除指定词条指定或全部回答
修改词条 [问题/下标] ?[下标/新回答] [新回答]：修改指定词条指定回答默认修改为第一条
查看词条 ?[问题/下标]：查看全部词条或对应词条回答
@@ -48,7 +49,7 @@ __plugin_cmd__ = [
"修改词条 [问题/下标] ?[下标/新回答] [新回答]",
"查看词条 ?[问题/下标]",
]
__plugin_version__ = 0.2
__plugin_version__ = 0.3
__plugin_author__ = "HibiKier & yajiwa"
__plugin_settings__ = {
"admin_level": Config.get_config("word_bank", "WORD_BANK_LEVEL [LEVEL]"),
@@ -87,21 +88,21 @@ async def _(bot: Bot, event: GroupMessageEvent, state: T_State, arg: Message = C
else:
_problem = problem
search_type = 0
if re.search("^关键字(.*)", msg):
if re.search("^关键字|词(.*)", msg):
search_type = 1
elif re.search("^模糊(.*)", msg):
search_type = 2
_builder = await get__builder(event, _problem, answer, idx)
if await _builder.save(search_type):
logger.info(f"已保存词条 问：{problem} 答：{msg}")
await add_word.send(f"已保存词条：{problem}")
logger.info(f"已保存词条 问：{_builder.problem} 答：{msg}")
await add_word.send("已保存词条：" + _builder.problem)
else:
await delete_word.send("保存失败，可能是回答重复")
@delete_word.handle()
async def _(event: GroupMessageEvent, arg: Message = CommandArg()):
msg = arg.extract_plain_text().strip()
msg = str(arg).strip()
if not msg:
await delete_word.finish("此命令之后需要跟随指定词条，通过“显示词条“查看")
index = None
@@ -119,17 +120,18 @@ async def _(event: GroupMessageEvent, arg: Message = CommandArg()):
if p:
problem = p[int(x)]
try:
_problem, problem = await get_one_problem(event, problem)
if answer := await WordBank.delete_problem_answer(
event.user_id, event.group_id, problem, index
event.user_id, event.group_id, _problem, index
):
await delete_word.send(f"删除词条成功：{problem}\n回答：\n{answer}")
await delete_word.send("删除词条成功：" + problem + f"\n回答：\n{answer}")
logger.info(
f"(USER {event.user_id}, GROUP "
f"{event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 删除词条: {problem}"
)
else:
await delete_word.send(f"删除词条：{problem} 失败，可能该词条不存在")
await delete_word.send("删除词条：" + problem + "失败，可能该词条不存在")
except IndexError:
await delete_word.send("指定下标错误...请通过查看词条来确定..")
@@ -165,21 +167,21 @@ async def _(bot: Bot, event: GroupMessageEvent, arg: Message = CommandArg()):
try:
if await _builder.update(index):
await update_word.send(f"修改词条成功：{problem}")
await update_word.send(f"修改词条成功：" + _builder.problem)
logger.info(
f"(USER {event.user_id}, GROUP "
f"{event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 修改词条: {problem}"
)
else:
await update_word.send(f"修改词条：{problem} 失败，可能该词条不存在")
await update_word.send(f"修改词条：" + _builder.problem + f"失败，可能该词条不存在")
except IndexError:
await update_word.send("指定下标错误...请通过查看词条来确定..")
@show_word.handle()
async def _(bot: Bot, event: GroupMessageEvent, arg: Message = CommandArg()):
msg = arg.extract_plain_text().strip()
msg = str(arg).strip()
if not msg:
_problem_list = await WordBank.get_group_all_problem(event.group_id)
if not _problem_list:
@@ -225,17 +227,28 @@ async def _(bot: Bot, event: GroupMessageEvent, arg: Message = CommandArg()):
)
)
else:
_answer_list = await WordBank.get_group_all_answer(event.group_id, msg)
_answer_list = []
if msg.startswith("id:"):
x = msg.split(":")[-1]
if not is_number(x) or int(x) < 0:
return await delete_word.finish("id必须为数字且符合规范！")
p = await WordBank.get_group_all_problem(event.group_id)
if p:
_problem = p[int(x)]
_answer_list = await WordBank.get_group_all_answer(event.group_id, _problem)
msg += '问' + await get_one_image_problem(event, _problem)
else:
_problem, msg = await get_one_problem(event, msg)
_answer_list = await WordBank.get_group_all_answer(event.group_id, _problem)
if not _answer_list:
await show_word.send("未收录该词条...")
else:
# 解析图片和@
_answer_img_nu_list = [await get_one_answer(event, format, answer, 0) for answer, format in _answer_list]
word_nu = len(_answer_img_nu_list)
img_nu = 0
answer = f"词条 {msg} 回答："
answer = "词条" + msg + "回答："
for i, x, in enumerate(_answer_img_nu_list):
r = re.findall(rf"\[CQ:image,file=", str(x))
if r:
@@ -254,15 +267,25 @@ async def _(bot: Bot, event: GroupMessageEvent, arg: Message = CommandArg()):
async def get__builder(event, _problem, answer, idx):
(data_dir / f"{event.group_id}").mkdir(exist_ok=True, parents=True)
(data_dir / f"{event.group_id}" / "problem").mkdir(exist_ok=True, parents=True)
_builder = WordBankBuilder(event.user_id, event.group_id, _problem)
problem = ''
_p = _problem
for at_ in get_message_at(event.json()):
r = re.search(rf"\[CQ:at,qq={at_}]", answer)
if r:
answer = answer.replace(f"[CQ:at,qq={at_}]", f"[__placeholder_{idx}]", 1)
_builder.set_placeholder(idx, at_)
idx += 1
r_problem = re.search(rf"\[CQ:at,qq={at_}]", _problem)
if r_problem:
q = await GroupInfoUser.get_member_info(
int(at_), event.group_id)
problem += _p[: _p.find(f"[CQ:at,qq={at_}]")] + "@" + q.user_name
_p = _p[_p.find(f"[CQ:at,qq={at_}]") + len(f"[CQ:at,qq={at_}]"):]
for img in get_message_img(event.json()):
_x = img.split("?")[0]
_x_list = img.split("?")
r = re.search(rf"\[CQ:image,file=(.*),url={_x}.*?]", answer)
if r:
rand = random.randint(1, 10000) + random.randint(1, 114514)
@@ -270,16 +293,26 @@ async def get__builder(event, _problem, answer, idx):
if f"__placeholder_{rand}_{idx}.jpg" not in os.listdir(data_dir / f"{event.group_id}"):
break
rand = random.randint(1, 10000) + random.randint(1, 114514)
for i in range(3):
answer = answer.replace(f",subType={i}", "")
answer = answer.replace(
rf"[CQ:image,file={r.group(1)},url={img}]",
f"[__placeholder_{idx}]",
)
strinfo = re.compile(f"\[CQ:image,file={r.group(1)},url={_x_list[0]}\?{_x_list[1]},subType=\d*?]")
answer = strinfo.sub(f"[__placeholder_{idx}]", answer)
await AsyncHttpx.download_file(
img, data_dir / f"{event.group_id}" / f"__placeholder_{rand}_{idx}.jpg"
)
_builder.set_placeholder(idx, f"__placeholder_{rand}_{idx}.jpg")
idx += 1
r_problem = re.search(rf"\[CQ:image,file=(.*?),url={_x}.*?]", _p)
if r_problem:
strinfo = re.compile(f",url={_x_list[0]}\?{_x_list[1]},subType=\d*?]")
_problem = strinfo.sub(f"]", _problem)
_p = strinfo.sub(f"]", _p)
problem += _p[: _p.find(f"[CQ:image,file={r_problem.group(1)}]")] + image(img)
_p = _p[_p.find(f"[CQ:image,file={r_problem.group(1)}]") + len(f"[CQ:image,file={r_problem.group(1)}]"):]
problem_img = r_problem.group(1)
if f"{problem_img}.jpg" not in os.listdir(data_dir / f"{event.group_id}" / f"problem"):
await AsyncHttpx.download_file(
img, data_dir / f"{event.group_id}" / f"problem" / f"{problem_img}.jpg"
)
_builder.set_answer(answer)
_builder.set_problem(_problem)
_builder.problem = problem + _p
return _builder
utils/utils.py
+25 -10
View File
@@ -16,7 +16,6 @@ try:
except ModuleNotFoundError:
import json
scheduler = require("nonebot_plugin_apscheduler").scheduler
@@ -108,8 +107,8 @@ class BanCheckLimiter:
self.mint[key] = 0
return False
if (
self.mint[key] >= self.default_count
and time.time() - self.mtime[key] < self.default_check_time
self.mint[key] >= self.default_count
and time.time() - self.mtime[key] < self.default_check_time
):
self.mtime[key] = time.time()
self.mint[key] = 0
@@ -203,10 +202,7 @@ def get_message_at(data: Union[str, Message]) -> List[int]:
for msg in data["message"]:
if msg["type"] == "at":
qq_list.append(int(msg["data"]["qq"]))
else:
for seg in data:
if seg.type == "image":
qq_list.append(seg.data["url"])
return qq_list
@@ -229,6 +225,25 @@ def get_message_img(data: Union[str, Message]) -> List[str]:
return img_list
def get_message_img_file(data: Union[str, Message]) -> List[str]:
"""
说明：
获取消息中所有的 图片file
参数：
:param data: event.json()
"""
file_list = []
if isinstance(data, str):
data = json.loads(data)
for msg in data["message"]:
if msg["type"] == "image":
file_list.append(msg["data"]["file"])
else:
for seg in data["image"]:
file_list.append(seg.data["file"])
return file_list
def get_message_text(data: Union[str, Message]) -> str:
"""
说明：
@@ -355,7 +370,7 @@ def cn2py(word: str) -> str:
def change_pixiv_image_links(
url: str, size: Optional[str] = None, nginx_url: Optional[str] = None
url: str, size: Optional[str] = None, nginx_url: Optional[str] = None
):
"""
说明：
@@ -373,7 +388,7 @@ def change_pixiv_image_links(
if nginx_url:
url = (
url.replace("i.pximg.net", nginx_url)
.replace("i.pixiv.cat", nginx_url)
.replace("_webp", "")
.replace("i.pixiv.cat", nginx_url)
.replace("_webp", "")
)
return url