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 7 Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main'

Browse Source
This commit is contained in:
HibiKier 2022-08-22 20:48:57 +08:00
commit afc05e1731
9 changed files with 74 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
plugins/genshin/query_user/bind/__init__.py
View File

@ -93,6 +93,8 @@ async def _(event: MessageEvent, cmd: Tuple[str, ...] = Command(), arg: Message
print(cookie)
cookie_json = json.loads(cookie)
print(cookie_json)
if 'login_ticket' not in cookie_json:
await bind.finish("请发送正确完整的cookie！")
login_ticket = cookie_json['login_ticket']
# try:
res = await AsyncHttpx.get(url=bbs_Cookie_url.format(login_ticket))

6
plugins/genshin/query_user/genshin_sign/__init__.py
View File

@ -44,6 +44,8 @@ async def _(event: MessageEvent, cmd: Tuple[str, ...] = Command()):
uid = await Genshin.get_user_uid(event.user_id)
if cmd == "查看我的cookie":
my_cookie = await Genshin.get_user_cookie(uid, True)
if isinstance(event, GroupMessageEvent):
await genshin_matcher.finish("请私聊查看您的cookie！")
await genshin_matcher.finish("您的cookie为" + my_cookie)
if not uid or not await Genshin.get_user_cookie(uid, True):
await genshin_matcher.finish("请先绑定uid和cookie！")
@ -51,9 +53,9 @@ async def _(event: MessageEvent, cmd: Tuple[str, ...] = Command()):
# await genshin_matcher.finish("请更新cookie！")
if cmd == "原神我硬签":
try:
await genshin_matcher.send("正在进行签到...", at_sender=True)
msg = await genshin_sign(uid)
return_data = await mihoyobbs_sign(event.user_id)
await genshin_matcher.send(return_data)
logger.info(
f"(USER {event.user_id}, "
f"GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'}) UID：{uid} 原神签到"
@ -70,7 +72,7 @@ async def _(event: MessageEvent, cmd: Tuple[str, ...] = Command()):
await u.clear_sign_time(uid)
next_date = await Genshin.random_sign_time(uid)
add_job(event.user_id, uid, next_date)
msg += f"\n因开启自动签到\n下一次签到时间为：{next_date.replace(microsecond=0)}"
msg += f"\n{return_data}\n因开启自动签到\n下一次签到时间为：{next_date.replace(microsecond=0)}"
except Exception as e:
msg = "原神签到失败..请尝试检查cookie或报告至管理员！"
logger.info(

8
plugins/genshin/query_user/mihoyobbs_sign/__init__.py
View File

@ -38,16 +38,18 @@ mihoyobbs_matcher = on_command(
@mihoyobbs_matcher.handle()
async def _(event: MessageEvent, cmd: Tuple[str, ...] = Command()):
await mihoyobbs_matcher.send("提交米游社签到申请")
await mihoyobbs_matcher.send("提交米游社签到申请", at_sender=True)
return_data = await mihoyobbs_sign(event.user_id)
if return_data:
await mihoyobbs_matcher.finish(return_data)
await mihoyobbs_matcher.finish(return_data, at_sender=True)
else:
await mihoyobbs_matcher.finish("米游社签到失败，请查看控制台输出")
await mihoyobbs_matcher.finish("米游社签到失败，请查看控制台输出", at_sender=True)
async def mihoyobbs_sign(user_id):
uid = await Genshin.get_user_uid(user_id)
if not uid or not await Genshin.get_user_cookie(uid, True):
await mihoyobbs_matcher.finish("请先绑定uid和cookie！", at_sender=True)
stuid = await Genshin.get_stuid(uid)
stoken = await Genshin.get_stoken(uid)
cookie = await Genshin.get_user_cookie(uid)

4
plugins/genshin/query_user/query_memo/data_source.py
View File

@ -35,8 +35,8 @@ async def _():
"20.png?x-oss-process=image//resize,s_600/quality,q_80/auto-orient,0/interlace,1/format,png",
"https://patchwiki.biligame.com/images/ys/thumb/c/cc/6k6kuj1kte6m1n7hexqfrn92z6h4yhh.png/60px-委托任务logo.png",
"https://patchwiki.biligame.com/images/ys/d/d9/t1hv6wpucbwucgkhjntmzroh90nmcdv.png",
"https://s3.bmp.ovh/imgs/2022/07/24/28d9338c7da4bcb2.png",
"https://genshin.honeyhunterworld.com/img/gadget/i_3016.png",
"https://s3.bmp.ovh/imgs/2022/08/21/3a3b2e6c22e305ff.png",
"https://s3.bmp.ovh/imgs/2022/08/21/c2d7ace21e1d46cf.png",
],
):
file = memo_path / name

2
plugins/quotations.py
View File

@ -24,7 +24,7 @@ __plugin_settings__ = {
}
quotations = on_regex("^[语录|二次元]$", priority=5, block=True)
quotations = on_regex("^(语录|二次元)$", priority=5, block=True)
url = "https://international.v1.hitokoto.cn/?c=a"

19
plugins/word_bank/_model.py
View File

@ -234,16 +234,27 @@ class WordBank(db.Model):
query = query.where(cls.word_scope == word_type)
sql_text += f" and word_scope = {word_scope}"
# 完全匹配
if await query.where(cls.problem == problem).gino.first():
if await query.where(
((cls.word_type == 0) | (cls.word_type == 3)) & (cls.problem == problem)
).gino.first():
return query.where(cls.problem == problem)
# 模糊匹配
if await query.where(cls.problem.contains(problem)).gino.first():
if await query.where(
(cls.word_type == 1) & (cls.problem.contains(problem))
).gino.first():
return query.where(cls.problem.contains(problem))
# 正则匹配
if await db.first(
db.text(sql_text + f" and word_type = 2 and word_scope != 999 and '{problem}' ~ problem;")
db.text(
sql_text
+ f" and word_type = 2 and word_scope != 999 and :problem ~ problem;"
),
problem=problem,
):
return sql_text + f" and word_type = 2 and word_scope != 999 and '{problem}' ~ problem;"
return (
sql_text
+ f" and word_type = 2 and word_scope != 999 and '{problem}' ~ problem;"
)
# if await db.first(
# db.text(sql_text + f" and word_type = 1 and word_scope != 999 and '{problem}' ~ problem;")
# ):

27
plugins/word_bank/_rule.py
View File

@ -1,31 +1,38 @@
import random
import imagehash
from PIL import Image
from io import BytesIO
from httpx import TimeoutException
from nonebot.typing import T_State
from nonebot.adapters.onebot.v11 import MessageEvent
from configs.path_config import TEMP_PATH
from utils.image_utils import get_img_hash
from utils.utils import get_message_text, get_message_img, get_message_at
from ._model import WordBank
from utils.http_utils import AsyncHttpx
async def check(event: MessageEvent) -> bool:
async def check(event: MessageEvent, state: T_State) -> bool:
text = get_message_text(event.message)
img = get_message_img(event.message)
at = get_message_at(event.message)
rand = random.randint(1, 100)
problem = text
if not text and len(img) == 1:
if await AsyncHttpx.download_file(img[0], TEMP_PATH / f"{event.user_id}_{rand}_word_bank_check.jpg"):
problem = str(get_img_hash(TEMP_PATH / f"{event.user_id}_{rand}_word_bank_check.jpg"))
try:
r = await AsyncHttpx.get(img[0])
problem = str(imagehash.average_hash(Image.open(BytesIO(r.content))))
except TimeoutException:
pass
if at:
temp = ''
for seg in event.message:
if seg.type == 'at':
temp += f"[at:{seg.data['qq']}]"
else:
elif isinstance(seg, str):
temp += seg
elif seg.type == 'text':
temp += seg.data["text"]
problem = temp
if problem:
return await WordBank.check(event, problem) is not None
if problem and (await WordBank.check(event, problem) is not None):
state["problem"] = problem
return True
return False

35
plugins/word_bank/message_handle.py
View File

@ -1,14 +1,10 @@
import random
from services import logger
from utils.image_utils import get_img_hash
from ._rule import check
from ._model import WordBank
from configs.path_config import DATA_PATH, TEMP_PATH
from configs.path_config import DATA_PATH
from nonebot.adapters.onebot.v11 import GroupMessageEvent, MessageEvent
from utils.utils import get_message_img, get_message_text, get_message_at
from nonebot import on_message
from utils.http_utils import AsyncHttpx
from nonebot.typing import T_State
__zx_plugin_name__ = "词库问答回复操作 [Hidden]"
@ -19,33 +15,12 @@ message_handle = on_message(priority=6, block=True, rule=check)
@message_handle.handle()
async def _(event: MessageEvent):
text = get_message_text(event.message)
img = get_message_img(event.message)
at = get_message_at(event.message)
problem = None
if not text and img and len(img) == 1:
rand = random.randint(1, 10000)
if await AsyncHttpx.download_file(img[0], TEMP_PATH / f"{event.user_id}_{rand}_word_bank.jpg"):
problem = str(get_img_hash(TEMP_PATH / f"{event.user_id}_{rand}_word_bank.jpg"))
elif at:
temp = ''
for seg in event.message:
if seg.type == 'at':
temp += f"[at:{seg.data['qq']}]"
else:
temp += seg
problem = temp
elif text:
problem = text
if problem:
async def _(event: MessageEvent, state: T_State):
if problem := state.get("problem"):
if msg := await WordBank.get_answer(event, problem):
await message_handle.send(msg)
logger.info(
f"(USER {event.user_id}, GROUP "
f"{event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 触发词条 {problem}"
)
)

32
plugins/word_bank/word_handle.py
View File

@ -1,3 +1,4 @@
import re
from typing import Tuple, Any, Optional
from nonebot.internal.params import Arg, ArgStr
@ -5,11 +6,12 @@ from nonebot.typing import T_State
from utils.utils import get_message_at, is_number, get_message_img
from nonebot.params import CommandArg, RegexGroup, Command
from nonebot.exception import FinishedException
from services.log import logger
from configs.path_config import DATA_PATH
from utils.message_builder import custom_forward_msg
from ._model import WordBank
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, Message, MessageEvent, PrivateMessageEvent
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, Message, MessageEvent, PrivateMessageEvent, unescape
from nonebot import on_command, on_regex
from configs.config import Config
from ._data_source import delete_word, update_word, show_word
@ -23,7 +25,7 @@ usage：
更推荐使用id方式删除
问题回答支持的CQ：at, face, image
查看词条命令：群聊时为 群词条+全局词条，私聊时为 私聊词条+全局词条
添加词条正则：添加词条(模糊|正则|图片)?问\s*?(\S*)\s*?答\s?(\S*)
添加词条正则：添加词条(模糊|正则|图片)?问\s*?(\S*\s?\S*)\s*?答\s?(\S*)
指令：
添加词条 ?[模糊|正则|图片]问...答...：添加问答词条，可重复添加相同问题的不同回答
删除词条 [问题/下标] ?[下标]：删除指定词条指定或全部回答
@ -47,7 +49,7 @@ __plugin_superuser_usage__ = r"""
usage:
在私聊中超级用户额外设置
指令：
(全局|私聊)?添加词条\s*?(模糊|正则|图片)?问\s*?(\S*)\s*?答\s?(\S*)：添加问答词条，可重复添加相同问题的不同回答
(全局|私聊)?添加词条\s*?(模糊|正则|图片)?问\s*?(\S*\s?\S*)\s*?答\s?(\S*)：添加问答词条，可重复添加相同问题的不同回答
全局添加词条
私聊添加词条
（私聊情况下）删除词条: 删除私聊词条
@ -74,7 +76,7 @@ data_dir = DATA_PATH / "word_bank"
data_dir.mkdir(parents=True, exist_ok=True)
add_word = on_regex(
r"^(全局|私聊)?添加词条\s*?(模糊|正则|图片)?问\s*?(\S*)\s*?答\s?(\S*)", priority=5, block=True
r"^(全局|私聊)?添加词条\s*?(模糊|正则|图片)?问\s*?(\S*\s?\S*)\s*?答\s?(\S*)", priority=5, block=True
)
delete_word_matcher = on_command("删除词条", aliases={'删除全局词条'}, priority=5, block=True)
@ -113,16 +115,21 @@ async def _(
if seg.type == 'text' and '答' in str(seg):
_problem = event.message[:index]
answer = event.message[index:]
answer[0] = str(answer[0])[str(answer).index('答')+1:]
_problem[0] = str(_problem[0])[str(_problem).index('问')+1:]
answer[0] = str(answer[0])[str(answer[0]).index('答')+1:]
_problem[0] = str(_problem[0])[str(_problem[0]).index('问')+1:]
if _problem[-1].type != 'at' or seg.data['text'][:seg.data['text'].index('答')].lstrip():
_problem.append(seg.data['text'][:seg.data['text'].index('答')])
temp = ''
for g in _problem:
if isinstance(g, str) or g.type == 'text':
if isinstance(g, str):
temp += g
elif g.type == 'text':
temp += g.data['text']
elif g.type == 'at':
temp += f"[at:{g.data['qq']}]"
problem = temp
break
problem = unescape(problem)
index = len((word_scope or "") + "添加词条" + (word_type or "") + problem) + 1
event.message[0] = event.message[0].data["text"][index + 1 :].strip()
state["word_scope"] = word_scope
@ -141,6 +148,11 @@ async def _(
problem_image: Message = Arg("problem_image"),
):
try:
if word_type == "正则":
try:
re.compile(problem)
except re.error:
await add_word.finish(f"添加词条失败，正则表达式 {problem} 非法！")
await WordBank.add_problem_answer(
event.user_id,
event.group_id if isinstance(event, GroupMessageEvent) and (not word_scope or word_scope == '1') else 0,
@ -150,6 +162,8 @@ async def _(
answer,
)
except Exception as e:
if isinstance(e, FinishedException):
await add_word.finish()
logger.error(
f"(USER {event.user_id}, GROUP "
f"{event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
@ -233,7 +247,7 @@ async def _(bot: Bot, event: GroupMessageEvent, arg: Message = CommandArg()):
not is_number(id_)
or int(id_) < 0
or int(id_)
> len(await WordBank.get_group_all_problem(event.group_id))
>= len(await WordBank.get_group_all_problem(event.group_id))
):
await show_word_matcher.finish("id必须为数字且在范围内")
id_ = int(id_)
@ -243,7 +257,7 @@ async def _(bot: Bot, event: GroupMessageEvent, arg: Message = CommandArg()):
not is_number(gid)
or int(gid) < 0
or int(gid)
> len(await WordBank.get_problem_by_scope(0))
>= len(await WordBank.get_problem_by_scope(0))
):
await show_word_matcher.finish("gid必须为数字且在范围内")
gid = int(gid)