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

Merge branch 'HibiKier:main' into main

Browse Source
This commit is contained in:
yajiwa
2022-05-29 21:03:12 +08:00
committed by GitHub
parent 374b7b7261 38501cebf1
commit e0412c7edb
28 changed files with 486 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+31 -4
View File
@@ -236,13 +236,40 @@ python bot.py
## 使用Docker
__Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
#### GitHub：[Sakuracio/zxenv](https://github.com/Sakuracio/zxenv)
#### GitHub：[Sakuracio/zhenxun_bot_docker](https://github.com/Sakuracio/zhenxun_bot_docker)
#### DockerHub：[hibikier/zhenxun_bot](https://hub.docker.com/r/hibikier/zhenxun_bot)
## 更新
### 2022/5/29 \[v0.1.5.8]
* 提供了真寻适配仓库的插件 安装/卸载 操作
* 暂时关闭了插件资源清空
* 通过指令安装插件时会在插件目录下生成plugin_info.json记录当前插件信息
### 2022/5/28
* 修复私聊无法添加昵称
* 修复原神玩家查询层岩巨渊地下矿区没开时报错
* 修复 ```休息吧``` 无法阻断戳一戳
* 当图库无图片时，戳一戳将略过发送图片
* 新增搜图提供配置项```ALLOW_GROUP_R18```：允许在群聊中使用r18参数
* 新增自动更新插件```UPDATE_REMIND```：新版本提醒，原配置项```AUTO_UPDATE_ZHENXUN```改为自动更新升级
* black_word新增当群权限为-1时不再检测该群
* 修复非超级用户绑定原神cookie会被black_word阻拦
* 修复webui中plugins2setting修改时会改变plugins2setting.cmd为字符串
* 修复微博热搜报错,发红包小bug [@pull/688](https://github.com/HibiKier/zhenxun_bot/pull/688)
* 更多的中文提示
### 2022/5/26
* 修复\[滴滴滴]会被转义成[滴滴滴]导致无法触发的问题
* 将一些错误以中文提示输出
* 更新BT搜索源地址 [@pull/668](https://github.com/HibiKier/zhenxun_bot/pull/668)
* 更新抽卡插件 [@pull/667](https://github.com/HibiKier/zhenxun_bot/pull/667)
### 2022/5/25
* 修复webui中CountLimit字段limit_type类型错误
@@ -267,9 +294,9 @@ __Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
* 添加pixiv搜图多关键词支持;修复p站搜图数量参数问题 [@pull/441](https://github.com/HibiKier/zhenxun_bot/pull/441)
* 修复开箱更新价格错误传参
* 修复pix无法正确查询uid
* 新增色图插件添加配置项ALLOW_GROUP_R18：允许群聊中使用色图r
* 新增PIX插件添加配置项ALLOW_GROUP_SETU：允许非超级用户使用-s参数
* 新增PIX插件添加配置项ALLOW_GROUP_R18：允许非超级用户使用-r参数
* 新增色图插件添加配置项```ALLOW_GROUP_R18```：允许群聊中使用色图r
* 新增PIX插件添加配置项```ALLOW_GROUP_SETU```：允许非超级用户使用-s参数
* 新增PIX插件添加配置项```ALLOW_GROUP_R18```：允许非超级用户使用-r参数
### 2022/5/22 \[v0.1.5.4]
__version__
+1 -1
View File
@@ -1 +1 @@
__version__: v0.1.5.6
__version__: v0.1.5.8
basic_plugins/hooks/auth_hook.py
+9 -2
View File
@@ -91,7 +91,10 @@ async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
except AttributeError:
pass
# 群黑名单检测 群总开关检测
if isinstance(event, GroupMessageEvent) or matcher.plugin_name == other_limit_plugins:
if (
isinstance(event, GroupMessageEvent)
or matcher.plugin_name in other_limit_plugins
):
try:
if (
group_manager.get_group_level(event.group_id) < 0
@@ -148,7 +151,11 @@ async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
raise IgnoredException("权限不足")
if module in plugins2info_dict.keys() and matcher.priority not in [1, 999]:
# 戳一戳单独判断
if isinstance(event, GroupMessageEvent) or isinstance(event, PokeNotifyEvent) or matcher.plugin_name in other_limit_plugins:
if (
isinstance(event, GroupMessageEvent)
or isinstance(event, PokeNotifyEvent)
or matcher.plugin_name in other_limit_plugins
):
if status_message_manager.get(event.group_id) is None:
status_message_manager.delete(event.group_id)
if plugins2info_dict[module]["level"] > group_manager.get_group_level(
basic_plugins/init_plugin_config/init_none_plugin_count_manager.py
+1 -1
View File
@@ -40,7 +40,7 @@ def init_none_plugin_count_manager():
plugins2block_manager.save()
plugins_manager.delete(module)
plugins_manager.save()
resources_manager.remove_resource(module)
# resources_manager.remove_resource(module)
none_plugin_count_manager.delete(module)
logger.info(f"{module}:{plugin_name} 插件疑似已删除，清除对应插件数据...")
except Exception as e:
basic_plugins/init_plugin_config/init_plugins_config.py
+25 -10
View File
@@ -3,6 +3,7 @@ from ruamel.yaml import round_trip_load, round_trip_dump, YAML
from utils.manager import admin_manager, plugins_manager
from configs.config import Config
from services.log import logger
from utils.text_utils import prompt2cn
from utils.utils import get_matchers
from ruamel import yaml
import nonebot
@@ -56,7 +57,9 @@ def init_plugins_config(data_path):
_override=True,
)
else:
Config.add_plugin_config(matcher.plugin_name, key, plugin_configs[key])
Config.add_plugin_config(
matcher.plugin_name, key, plugin_configs[key]
)
else:
plugin_configs = _data[matcher.plugin_name]
for key in plugin_configs:
@@ -79,7 +82,10 @@ def init_plugins_config(data_path):
_data[plugin].yaml_set_start_comment(plugin_name, indent=2)
# 初始化未设置的管理员权限等级
for k, v in Config.get_admin_level_data():
admin_manager.set_admin_level(k, v)
try:
admin_manager.set_admin_level(k, v)
except KeyError as e:
raise KeyError(f"{e} ****** 请检查是否有插件加载失败 ******")
# 存完插件基本设置
with open(plugins2config_file, "w", encoding="utf8") as wf:
round_trip_dump(
@@ -96,14 +102,23 @@ def init_plugins_config(data_path):
for plugin in Config.keys():
_tmp_data[plugin] = {}
for k in Config[plugin].keys():
if _data.get(plugin) and k in _data[plugin].keys():
Config.set_config(plugin, k, _data[plugin][k])
if level2module := Config.get_level2module(plugin, k):
try:
admin_manager.set_admin_level(level2module, _data[plugin][k])
except KeyError:
logger.warning(f"{level2module} 设置权限等级失败：{_data[plugin][k]}")
_tmp_data[plugin][k] = Config.get_config(plugin, k)
try:
if _data.get(plugin) and k in _data[plugin].keys():
Config.set_config(plugin, k, _data[plugin][k])
if level2module := Config.get_level2module(plugin, k):
try:
admin_manager.set_admin_level(
level2module, _data[plugin][k]
)
except KeyError:
logger.warning(
f"{level2module} 设置权限等级失败：{_data[plugin][k]}"
)
_tmp_data[plugin][k] = Config.get_config(plugin, k)
except AttributeError as e:
raise AttributeError(
f"{e}\n" + prompt2cn("可能为config.yaml配置文件填写不规范", 46)
)
Config.save()
temp_file = Path() / "configs" / "temp_config.yaml"
try:
basic_plugins/nickname.py
+2 -2
View File
@@ -1,4 +1,4 @@
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, PrivateMessageEvent, Message
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, PrivateMessageEvent, Message, MessageEvent
from nonebot import on_command
from nonebot.typing import T_State
from nonebot.rule import to_me
@@ -47,7 +47,7 @@ cancel_nickname = on_command("取消昵称", rule=to_me(), priority=5, block=Tru
@nickname.handle()
async def _(bot: Bot, event: GroupMessageEvent, arg: Message = CommandArg()):
async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
msg = arg.extract_plain_text().strip()
if not msg:
await nickname.finish("叫你空白？叫你虚空？叫你无名？？", at_sender=True)
basic_plugins/plugin_shop/__init__.py
+68
View File
@@ -0,0 +1,68 @@
from nonebot import on_command, on_regex
from .data_source import show_plugin_repo, install_plugin, uninstall_plugin, download_json
from services.log import logger
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, Message, MessageEvent
from nonebot.params import CommandArg
from utils.message_builder import image
from nonebot.permission import SUPERUSER
__zx_plugin_name__ = "插件商店 [Superuser]"
__plugin_usage__ = """
usage：
下载安装插件
指令：
查看插件仓库
更新插件仓库
安装插件 [name/id]
卸载插件 [name/id]
""".strip()
__plugin_des__ = "从真寻适配仓库中下载插件"
__plugin_cmd__ = [""]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
show_repo = on_regex("^查看插件仓库$", priority=1, block=True, permission=SUPERUSER)
update_repo = on_regex("^更新插件仓库$", priority=1, block=True, permission=SUPERUSER)
install_plugin_matcher = on_command("安装插件", priority=1, block=True, permission=SUPERUSER)
uninstall_plugin_matcher = on_command("卸载插件", priority=1, block=True, permission=SUPERUSER)
@install_plugin_matcher.handle()
async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
arg = arg.extract_plain_text().strip()
msg = await install_plugin(arg)
await install_plugin_matcher.send(msg)
@uninstall_plugin_matcher.handle()
async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
arg = arg.extract_plain_text().strip()
msg = await uninstall_plugin(arg)
await install_plugin_matcher.send(msg)
@update_repo.handle()
async def _(bot: Bot, event: MessageEvent):
code = await download_json()
if code == 200:
await update_repo.finish("更新插件仓库信息成功！")
await update_repo.send("更新插件仓库信息失败！")
@show_repo.handle()
async def _(bot: Bot, event: MessageEvent):
msg = await show_plugin_repo()
if isinstance(msg, int):
await show_repo.finish("文件下载失败或解压失败..")
await show_repo.send(image(b64=msg))
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 查看插件仓库"
)
basic_plugins/plugin_shop/data_source.py
+185
View File
@@ -0,0 +1,185 @@
import os
import shutil
import zipfile
from pathlib import Path
from typing import Union, Tuple
import ujson as json
from services import logger
from utils.http_utils import AsyncHttpx
from configs.path_config import TEMP_PATH, DATA_PATH
from utils.image_utils import text2image, BuildImage
from utils.utils import is_number
path = DATA_PATH / "plugin_shop"
if not path.exists():
path.mkdir(parents=True, exist_ok=True)
repo_json_url = "https://github.com/zhenxun-org/nonebot_plugins_zhenxun_bot/archive/refs/heads/index.zip"
repo_zip_path = TEMP_PATH / "plugin_repo_json.zip"
plugin_json = path / "zhenxun_plugins.json"
extensive_plugin_path = Path() / "extensive_plugin"
path = DATA_PATH / "plugin_shop"
if not path.exists():
path.mkdir(parents=True, exist_ok=True)
if not extensive_plugin_path.exists():
extensive_plugin_path.mkdir(parents=True, exist_ok=True)
data = {}
async def install_plugin(name: str) -> str:
"""
安装插件
:param name: 插件名或下标
"""
try:
if is_number(name):
name, code = await get_plugin_name(int(name))
if code != 200:
return name
plugin_url = data[name]["download_url"]
url = (await AsyncHttpx.get(plugin_url)).headers.get("Location")
zip_file = TEMP_PATH / f"{name}.zip"
if zip_file.exists():
zip_file.unlink()
if await AsyncHttpx.download_file(url, zip_file):
logger.debug("开始解压插件压缩包...")
# 解压
zf = zipfile.ZipFile(zip_file, "r")
extract_path = TEMP_PATH / f"{name}"
if extract_path.exists():
shutil.rmtree(extract_path.absolute(), ignore_errors=True)
extract_path.mkdir(exist_ok=True, parents=True)
for file in zf.namelist():
zf.extract(file, extract_path)
zf.close()
logger.debug("解压插件压缩包完成...")
logger.debug("开始移动插件文件夹...")
if (extensive_plugin_path / f"{name}").exists():
logger.debug("extensive_plugin目录下文件夹已存在，删除该目录插件文件夹...")
shutil.rmtree(
(extensive_plugin_path / f"{name}").absolute(), ignore_errors=True
)
extract_path.rename(extensive_plugin_path / f"{name}")
tmp = ""
if "pyproject.toml" in os.listdir(extensive_plugin_path / f"{name}"):
tmp = "检测到该插件含有额外依赖，当前安装无法保证依赖完全安装成功。"
os.system(
f"poetry run pip install -r {(extensive_plugin_path / f'{name}' / 'pyproject.toml').absolute()}"
)
with open(extensive_plugin_path / f'{name}' / "plugin_info.json", 'w') as f:
json.dump(data[name], f, ensure_ascii=False, indent=4)
logger.debug("移动插件文件夹完成...")
logger.info(f"成功安装插件 {name} 成功！\n{tmp}")
return f"成功安装插件 {name}，请重启bot！"
except Exception as e:
logger.error(f"安装插件 {name} 失败 {type(e)}：{e}")
return f"安装插件 {name} 失败 {type(e)}：{e}"
async def uninstall_plugin(name: str) -> str:
"""
删除插件
:param name: 插件名或下标
"""
try:
if is_number(name):
name, code = await get_plugin_name(int(name))
if code != 200:
return name
if name not in os.listdir(extensive_plugin_path):
return f"未安装 {name} 插件！"
shutil.rmtree((extensive_plugin_path / name).absolute(), ignore_errors=True)
logger.info(f"插件 {name} 删除成功！")
return f"插件 {name} 删除成功！"
except Exception as e:
logger.error(f"删除插件 {name} 失败 {type(e)}：{e}")
return f"删除插件 {name} 失败 {type(e)}：{e}"
async def show_plugin_repo() -> Union[int, str]:
"""
获取插件仓库数据并格式化
"""
if not plugin_json.exists():
code = await download_json()
if code != 200:
return code
plugin_info = json.load(open(plugin_json, "r", encoding="utf8"))
image_list = []
w, h = 0, 0
line_height = 10
for i, key in enumerate(plugin_info.keys()):
data[key] = {
"名称": plugin_info[key]["plugin_name"],
"模块": key,
"作者": plugin_info[key]["author"],
"版本": plugin_info[key]["version"],
"简介": plugin_info[key]["introduction"],
"download_url": plugin_info[key]["download_url"],
"github_url": plugin_info[key]["github_url"],
}
s = (
f'id：{i+1}\n名称：{plugin_info[key]["plugin_name"]}\n'
f'模块：{key}\n'
f'作者：{plugin_info[key]["author"]}\n'
f'版本：{plugin_info[key]["version"]}\n'
f'简介：{plugin_info[key]["introduction"]}\n'
f"-------------------"
)
img = await text2image(s, font_size=20, color="#f9f6f2")
w = w if w > img.w else img.w
h += img.h + line_height
image_list.append(img)
A = BuildImage(w + 50, h + 50, color="#f9f6f2")
cur_h = 25
for img in image_list:
await A.apaste(img, (25, cur_h))
cur_h += img.h + line_height
return A.pic2bs4()
async def download_json() -> int:
"""
下载插件库json文件
"""
try:
url = (await AsyncHttpx.get(repo_json_url)).headers.get("Location")
if repo_zip_path.exists():
repo_zip_path.unlink()
if await AsyncHttpx.download_file(url, repo_zip_path):
zf = zipfile.ZipFile(repo_zip_path, "r")
extract_path = path / "temp"
for file in zf.namelist():
zf.extract(file, extract_path)
zf.close()
if plugin_json.exists():
plugin_json.unlink()
(
extract_path
/ "nonebot_plugins_zhenxun_bot-index"
/ "zhenxun_plugins.json"
).rename(plugin_json)
shutil.rmtree(extract_path.absolute(), ignore_errors=True)
return 200
except Exception as e:
logger.error(f"下载插件库压缩包失败或解压失败 {type(e)}：{e}")
return 999
async def get_plugin_name(name: int) -> Tuple[str, int]:
"""
通过下标获取插件名
:param name: 下标
"""
name = int(name)
if not data:
await show_plugin_repo()
if name < 1 or name > len(data.keys()):
return "下标超过上下限！", 999
name = list(data.keys())[name - 1]
return name, 200
bot.py
+1
View File
@@ -12,6 +12,7 @@ driver.on_shutdown(disconnect)
nonebot.load_plugin("nonebot_plugin_apscheduler")
nonebot.load_plugins("basic_plugins")
nonebot.load_plugins("plugins")
nonebot.load_plugins("extensive_plugin")
# 最后加载权限控制
nonebot.load_plugins("basic_plugins/hooks")
configs/utils/__init__.py
+30 -5
View File
@@ -2,6 +2,7 @@ from typing import Optional, Any, Union
from pathlib import Path
from ruamel.yaml import YAML
from ruamel import yaml
from ruamel.yaml.scanner import ScannerError
class ConfigsManager:
@@ -21,9 +22,24 @@ class ConfigsManager:
if file.exists():
with open(file, "r", encoding="utf8") as f:
self._data = _yaml.load(f)
if not self._data:
self.file.unlink()
raise ValueError(
"配置文件为空！\n"
"***********************************************************\n"
"****** 配置文件 plugins2config.yaml 为空，已删除，请重启 ******\n"
"***********************************************************"
)
if self._simple_file.exists():
with open(self._simple_file, "r", encoding="utf8") as f:
self._simple_data = _yaml.load(f)
try:
with open(self._simple_file, "r", encoding="utf8") as f:
self._simple_data = _yaml.load(f)
except ScannerError as e:
raise ScannerError(
f"{e}\n**********************************************\n"
f"****** 可能为config.yaml配置文件填写不规范 ******\n"
f"**********************************************"
)
def add_plugin_config(
self,
@@ -86,7 +102,10 @@ class ConfigsManager:
:param value: 值
"""
if module in self._data.keys():
if self._data[module].get(key) is not None and self._data[module][key] != value:
if (
self._data[module].get(key) is not None
and self._data[module][key] != value
):
self._data[module][key]["value"] = value
self._simple_data[module][key] = value
self.save()
@@ -115,7 +134,9 @@ class ConfigsManager:
self._data[module][key]["default_value"] = value
self.save()
def get_config(self, module: str, key: str, default: Optional[Any] = None) -> Optional[Any]:
def get_config(
self, module: str, key: str, default: Optional[Any] = None
) -> Optional[Any]:
"""
获取指定配置值
:param module: 模块名
@@ -161,7 +182,11 @@ class ConfigsManager:
if save_simple_data:
with open(self._simple_file, "w", encoding="utf8") as f:
yaml.dump(
self._simple_data, f, indent=2, Dumper=yaml.RoundTripDumper, allow_unicode=True
self._simple_data,
f,
indent=2,
Dumper=yaml.RoundTripDumper,
allow_unicode=True,
)
path = path if path else self.file
with open(path, "w", encoding="utf8") as f:
plugins/black_word/__init__.py
+5
View File
@@ -8,6 +8,7 @@ from nonebot.adapters.onebot.v11 import (
from nonebot.matcher import Matcher
from nonebot.message import run_preprocessor
from utils.image_utils import BuildImage
from utils.manager import group_manager
from utils.utils import get_message_text, is_number
from nonebot.params import RegexGroup, CommandArg
from .utils import black_word_manager
@@ -155,7 +156,11 @@ async def _(
and matcher.plugin_name == "black_word"
and not await BanUser.is_ban(event.user_id)
and str(event.user_id) not in bot.config.superusers
and not get_message_text(event.json()).startswith("原神绑定")
):
# 屏蔽群权限-1的群
if isinstance(event, GroupMessageEvent) and group_manager.get_group_level(event.group_id) < 0:
return
user_id = event.user_id
group_id = event.group_id if isinstance(event, GroupMessageEvent) else None
msg = get_message_text(event.json())
plugins/bt/data_source.py
+2 -2
View File
@@ -9,7 +9,7 @@ import platform
# asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
url = "http://www.eclzz.mobi"
url = "http://www.eclzz.win"
async def get_bt_info(keyword: str, page: int):
@@ -19,7 +19,7 @@ async def get_bt_info(keyword: str, page: int):
:param page: 页数
"""
text = (await AsyncHttpx.get(f"{url}/s/{keyword}_rel_{page}.html", timeout=5)).text
if text.find("大约0条结果") != -1:
if "大约0条结果" in text:
return
soup = BeautifulSoup(text, "lxml")
item_lst = soup.find_all("div", {"class": "search-item"})
plugins/check_zhenxun_update/__init__.py
+22 -16
View File
@@ -25,6 +25,11 @@ __plugin_cmd__ = ["检查更新真寻", "重启"]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_configs__ = {
"UPDATE_REMIND": {
"value": True,
"help": "真寻是否检测版本状态",
"default": True,
},
"AUTO_UPDATE_ZHENXUN": {
"value": False,
"help": "真寻是否自动检查更新",
@@ -72,7 +77,7 @@ async def _():
await restart.finish("暂无windows重启脚本...")
@restart.got("flag", prompt="确定是否重启真寻？（重启失败咱们将失去联系，请谨慎！）")
@restart.got("flag", prompt="确定是否重启真寻？确定请回复[是|好|确定]（重启失败咱们将失去联系，请谨慎！）")
async def _(flag: str = ArgStr("flag")):
if flag.lower() in ["true", "是", "好", "确定", "确定是"]:
await restart.send("开始重启真寻..请稍等...")
@@ -88,7 +93,7 @@ async def _(flag: str = ArgStr("flag")):
minute=0,
)
async def _():
if Config.get_config("check_zhenxun_update", "AUTO_UPDATE_ZHENXUN"):
if Config.get_config("check_zhenxun_update", "UPDATE_REMIND"):
_version = "v0.0.0"
_version_file = Path() / "__version__"
if _version_file.exists():
@@ -108,17 +113,18 @@ async def _():
message=f"检测到真寻版本更新\n"
f"当前版本：{_version}，最新版本：{latest_version}",
)
# try:
# code = await check_update(bot)
# except Exception as e:
# logger.error(f"更新真寻未知错误 {type(e)}：{e}")
# await bot.send_private_msg(
# user_id=int(list(bot.config.superusers)[0]),
# message=f"更新真寻未知错误 {type(e)}：{e}\n",
# )
# else:
# if code == 200:
# await bot.send_private_msg(
# user_id=int(list(bot.config.superusers)[0]),
# message=f"更新完毕，请重启真寻....",
# )
if Config.get_config("check_zhenxun_update", "AUTO_UPDATE_ZHENXUN"):
try:
code = await check_update(bot)
except Exception as e:
logger.error(f"更新真寻未知错误 {type(e)}：{e}")
await bot.send_private_msg(
user_id=int(list(bot.config.superusers)[0]),
message=f"更新真寻未知错误 {type(e)}：{e}\n",
)
else:
if code == 200:
await bot.send_private_msg(
user_id=int(list(bot.config.superusers)[0]),
message=f"更新完毕，请重启真寻....",
)
plugins/dialogue/__init__.py
+1 -1
View File
@@ -53,7 +53,7 @@ __plugin_settings__ = {
dialogue_data = {}
dialogue = on_command("[滴滴滴]", aliases={"滴滴滴-"}, priority=5, block=True)
dialogue = on_command("&#91;滴滴滴&#93;", aliases={"滴滴滴-"}, priority=5, block=True)
reply = on_command("/t", priority=1, permission=SUPERUSER, block=True)
plugins/draw_card/handles/prts_handle.py
+43 -31
View File
@@ -3,7 +3,6 @@ import random
import dateparser
from lxml import etree
from PIL import ImageDraw
from bs4 import BeautifulSoup
from datetime import datetime
from urllib.parse import unquote
from typing import List, Optional, Tuple
@@ -19,7 +18,7 @@ except ModuleNotFoundError:
import json
from .base_handle import BaseHandle, BaseData, UpChar, UpEvent
from ..config import draw_config, DRAW_PATH
from ..config import draw_config
from ..util import remove_prohibited_str, cn2py, load_font
from utils.image_utils import BuildImage
@@ -32,8 +31,9 @@ class Operator(BaseData):
class PrtsHandle(BaseHandle[Operator]):
def __init__(self):
super().__init__("prts", "明日方舟", "#eff2f5")
super().__init__(game_name="prts", game_name_cn="明日方舟")
self.max_star = 6
self.game_card_color = "#eff2f5"
self.config = draw_config.prts
self.ALL_OPERATOR: List[Operator] = []
@@ -41,8 +41,8 @@ class PrtsHandle(BaseHandle[Operator]):
def get_card(self, add: float) -> Operator:
star = self.get_star(
[6, 5, 4, 3],
[
star_list=[6, 5, 4, 3],
probability_list=[
self.config.PRTS_SIX_P + add,
self.config.PRTS_FIVE_P,
self.config.PRTS_FOUR_P,
@@ -74,7 +74,7 @@ class PrtsHandle(BaseHandle[Operator]):
acquire_operator = random.choice(all_operators)
return acquire_operator
def get_cards(self, count: int) -> List[Tuple[Operator, int]]:
def get_cards(self, count: int, **kwargs) -> List[Tuple[Operator, int]]:
card_list = [] # 获取所有角色
add = 0.0
count_idx = 0
@@ -162,8 +162,11 @@ class PrtsHandle(BaseHandle[Operator]):
def load_up_char(self):
try:
if (DRAW_PATH / "draw_card_up" / f"{self.game_name}_up_char.json").exists():
data = self.load_data(f"draw_card_up/{self.game_name}_up_char.json")
data = self.load_data(f"draw_card_up/{self.game_name}_up_char.json")
"""这里的 waring 有点模糊，更新游戏信息时没有up池的情况下也会报错，所以细分了一下"""
if not data:
logger.warning(f"当前无UP池或 {self.game_name}_up_char.json 文件不存在")
else:
self.UP_EVENT = UpEvent.parse_obj(data.get("char", {}))
except ValidationError:
logger.warning(f"{self.game_name}_up_char 解析出错")
@@ -211,6 +214,7 @@ class PrtsHandle(BaseHandle[Operator]):
await self.update_up_char()
async def update_up_char(self):
"""重载卡池"""
announcement_url = "https://ak.hypergryph.com/news.html"
result = await self.get_url(announcement_url)
if not result:
@@ -224,45 +228,53 @@ class PrtsHandle(BaseHandle[Operator]):
end_time = None
up_chars = []
pool_img = ""
title = ""
for activity_url in activity_urls:
for activity_url in activity_urls[:10]: # 减少响应时间, 10个就够了
activity_url = f"https://ak.hypergryph.com{activity_url}"
result = await self.get_url(activity_url)
if not result:
logger.warning(f"{self.game_name_cn}获取公告 {activity_url} 出错")
continue
soup = BeautifulSoup(result, "lxml")
contents = soup.find_all("p")
"""因为鹰角的前端太自由了，这里重写了匹配规则以尽可能避免因为前端乱七八糟而导致的重载失败"""
dom = etree.HTML(result, etree.HTMLParser())
contents = dom.xpath(
"//div[@class='article-content']/p/text() | //div[@class='article-content']/p/span/text() | //div[@class='article-content']/div[@class='media-wrap image-wrap']/img/@src"
)
title = ""
time = ""
chars: List[str] = []
for index, content in enumerate(contents):
if re.search("(.*)(寻访|复刻).*?开启", content.text):
title = content.text
if "【" in title and "】" in title:
title = re.split(r"[【】]", title)[1]
lines = [str(contents[index + i + 1].text) for i in range(5)]
time = ""
chars: List[str] = []
for line in lines:
if re.search("(.*)(寻访|复刻).*?开启", content):
title = re.split(r"[【】]", content)
title = "".join(title[1:-1]) if "-" in title else title[1]
lines = [contents[index-2+_] for _ in range(8)] # 从 -2 开始是因为xpath获取的时间有的会在寻访开启这一句之前
lines.append("") # 防止IndexError，加个空字符串
for idx, line in enumerate(lines):
match = re.search(
r"(\d{1,2}月\d{1,2}日.*?-.*?\d{1,2}月\d{1,2}日.*?$)", line
)
if match:
time = match.group(1)
if "★" in line:
"""这里修复了某些池子六星名称显示错误的问题(如奔崖号角)"""
if line[0] != '★':
idx = line.find('★')
line = line[idx:]
chars.append(line)
"""因为 <p> 的诡异排版，所以有了下面的一段"""
if ("★★" in line and "%" in line) or ("★★" in line and "%" in lines[idx + 1]):
chars.append(line) if ("★★" in line and "%" in line) else chars.append(line + lines[idx + 1])
if not time:
continue
start, end = time.replace("月", "/").replace("日", "").split("-")[:2]
start, end = time.replace("月", "/").replace("日", " ").split("-")[:2] # 日替换为空格是因为有日后面不接空格的情况，导致 split 出问题
start_time = dateparser.parse(start)
end_time = dateparser.parse(end)
pool_img = content.find_previous("img")["src"]
pool_img = contents[index-2]
r"""两类格式：用/分割，用\分割；★+(概率)+名字，★+名字+(概率)"""
for char in chars:
star = char.split("（")[0].count("★")
name = re.split(r"[：（]", char)[1]
names = name.split("/") if "/" in name else [name]
name = re.split(r"[：（]", char)[1] if "★（" not in char else re.split("）：", char)[1] # 有的括号在前面有的在后面
if "\\" in name:
names = name.split("\\")
elif "/" in name:
names = name.split("/")
else:
names = [name] # 既有用/分割的，又有用\分割的
names = [name.replace("[限定]", "").strip() for name in names]
if "权值" in char:
match = re.search(r"（在.*?以.*?(\d+).*?倍权值.*?）", char)
@@ -276,7 +288,7 @@ class PrtsHandle(BaseHandle[Operator]):
up_chars.append(
UpChar(name=name, star=star, limited=False, zoom=zoom)
)
break
break # 这里break会导致个问题：如果一个公告里有两个池子，会漏掉下面的池子，比如 5.19 的定向寻访。但目前我也没啥好想法解决
if title and start_time and end_time:
if start_time <= datetime.now() <= end_time:
self.UP_EVENT = UpEvent(
plugins/genshin/query_user/query_role/draw_image.py
+13 -12
View File
@@ -330,18 +330,19 @@ def get_country_data_image(world_data_dict: Dict) -> BuildImage:
f"{world_data_dict['层岩巨渊']['exploration_percentage'] / 10}%",
fill=(255, 255, 255),
)
content_bk.text((300, 85), "地下矿区探索", fill=(239, 211, 114))
content_bk.text(
(570, 85),
f"{world_data_dict['层岩巨渊·地下矿区']['exploration_percentage'] / 10}%",
fill=(255, 255, 255),
)
content_bk.text((300, 150), "流明石触媒", fill=(239, 211, 114))
content_bk.text(
(570, 150),
f"LV.{world_data_dict['层岩巨渊·地下矿区']['offerings'][0]['level']}",
fill=(255, 255, 255),
)
if world_data_dict.get('层岩巨渊·地下矿区'):
content_bk.text((300, 85), "地下矿区探索", fill=(239, 211, 114))
content_bk.text(
(570, 85),
f"{world_data_dict['层岩巨渊·地下矿区']['exploration_percentage'] / 10}%",
fill=(255, 255, 255),
)
content_bk.text((300, 150), "流明石触媒", fill=(239, 211, 114))
content_bk.text(
(570, 150),
f"LV.{world_data_dict['层岩巨渊·地下矿区']['offerings'][0]['level']}",
fill=(255, 255, 255),
)
elif country in ["龙脊雪山"]:
content_bk.text((300, 40), "探索", fill=(239, 211, 114))
content_bk.text(
plugins/gold_redbag/data_source.py
+3 -3
View File
@@ -64,8 +64,8 @@ async def _generate_open_redbag_pic(user_id: int, send_user_nickname: str, amoun
head = BuildImage(1000, 980, font_size=30, background=f'{IMAGE_PATH}/prts/redbag_12.png')
size = BuildImage(0, 0, font_size=50).getsize(send_user_nickname)
# QQ头像
ava_bk = BuildImage(100 + size[0], 66, color='white', font_size=50)
ava = BuildImage(66, 66, background=BytesIO(await get_user_avatar(user_id)))
ava_bk = BuildImage(100 + size[0], 66, is_alpha=True, font_size=50)
ava = BuildImage(66, 66, is_alpha=True, background=BytesIO(await get_user_avatar(user_id)))
ava_bk.paste(ava)
ava_bk.text((100, 7), send_user_nickname)
# ava_bk.show()
@@ -73,7 +73,7 @@ async def _generate_open_redbag_pic(user_id: int, send_user_nickname: str, amoun
head.paste(ava_bk, (int((1000 - ava_bk_w) / 2), 300))
# 金额
size = BuildImage(0, 0, font_size=150).getsize(amount)
price = BuildImage(size[0], size[1], font_size=150)
price = BuildImage(size[0], size[1], is_alpha=True, font_size=150)
price.text((0, 0), amount, fill=(209, 171, 108))
# 金币中文
head.paste(price, (int((1000 - size[0]) / 2) - 50, 460))
plugins/image_management/send_image/__init__.py
+1 -1
View File
@@ -36,7 +36,7 @@ __plugin_settings__ = {
"limit_superuser": False,
"cmd": ["发送图片"] + Config.get_config("image_management", "IMAGE_DIR_LIST"),
}
__plugin_resources__ = {"pa": IMAGE_PATH}
__plugin_resources__ = {"pa": IMAGE_PATH / "pa"}
Config.add_plugin_config(
"_task",
plugins/open_cases/__init__.py
+2 -2
View File
@@ -212,7 +212,7 @@ update_price = on_command("更新开箱价格", priority=1, permission=SUPERUSER
@update_price.handle()
async def _(event: MessageEvent, arg: Message = CommandArg()):
await update_price.send(await util_get_buff_price(arg.extract_plain_text().strip()))
await update_price.send(await util_get_buff_price(arg.extract_plain_text().strip() or "狂牙大行动"))
update_img = on_command("更新开箱图片", priority=1, permission=SUPERUSER, block=True)
@@ -220,7 +220,7 @@ update_img = on_command("更新开箱图片", priority=1, permission=SUPERUSER,
@update_img.handle()
async def _(event: MessageEvent, arg: Message = CommandArg()):
await update_img.send(await util_get_buff_img(str(arg.extract_plain_text().strip())))
await update_img.send(await util_get_buff_img(arg.extract_plain_text().strip() or "狂牙大行动"))
# 重置开箱
plugins/pixiv_rank_search/__init__.py
+6 -1
View File
@@ -71,6 +71,11 @@ __plugin_configs__ = {
"value": 20,
"help": "作品最大页数限制，超过的作品会被略过",
"default_value": 20
},
"ALLOW_GROUP_R18": {
"value": False,
"help": "允许群聊中使用 r18 参数",
"default_value": False
}
}
Config.add_plugin_config(
@@ -154,7 +159,7 @@ async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
msg = arg.extract_plain_text().strip()
if isinstance(event, GroupMessageEvent):
if "r18" in msg.lower():
if "r18" in msg.lower() and not Config.get_config("pixiv_rank_search", "ALLOW_GROUP_R18"):
await pixiv_keyword.finish("(脸红#) 你不会害羞的 八嘎！", at_sender=True)
r18 = 0 if "r18" in msg else 1
msg = msg.replace("r18", "").strip().split()
plugins/poke/__init__.py
+3 -3
View File
@@ -62,12 +62,12 @@ async def _poke_event(event: PokeNotifyEvent):
rst = "气死我了！"
await poke_.finish(rst + random.choice(poke__reply), at_sender=True)
rand = random.random()
if rand <= 0.3:
path = random.choice(["luoli", "meitu"])
path = random.choice(["luoli", "meitu"])
if rand <= 0.3 and len(os.listdir(IMAGE_PATH / "image_management" / path)) > 0:
index = random.randint(0, len(os.listdir(IMAGE_PATH / "image_management" / path)))
result = f"id：{index}" + image(f"{index}.jpg", "image_management/" + path)
await poke_.send(result)
logger.info(f"USER {event.user_id} 戳了戳我 回复: {result} \n {result}")
logger.info(f"USER {event.user_id} 戳了戳我 回复: {result} {result}")
elif 0.3 < rand < 0.6:
voice = random.choice(os.listdir(RECORD_PATH / "dinggong"))
result = record(voice, "dinggong")
plugins/wbtop/data_source.py
+1 -1
View File
@@ -52,7 +52,7 @@ def gen_wbtop_pic(data: dict) -> MessageSegment:
text_bk = BuildImage(700, 32 * 50, 700, 32, color="#797979")
for i, data in enumerate(data):
title = f"{i + 1}. {data['hot_word']}"
hot = data["hot_word_num"]
hot = str(data["hot_word_num"])
img = BuildImage(700, 30, font_size=20)
w, h = img.getsize(title)
img.text((10, int((30 - h) / 2)), title)
plugins/web_ui/api/plugins.py
+2
View File
@@ -127,6 +127,8 @@ def _(plugin: Plugin, user: User = Depends(token_to_user)) -> Result:
else:
if plugin.plugin_settings:
for key, value in plugin.plugin_settings:
if plugin.plugin_settings.cmd:
plugin.plugin_settings.cmd = plugin.plugin_settings.cmd.split(',')
plugins2settings_manager.set_module_data(plugin.model, key, value)
if plugin.plugin_manager:
for key, value in plugin.plugin_manager:
services/db_context.py
+3
View File
@@ -1,6 +1,7 @@
from gino import Gino
from .log import logger
from utils.text_utils import prompt2cn
from configs.config import bind, sql_name, user, password, address, port, database
@@ -9,6 +10,8 @@ db = Gino()
async def init():
if not bind and (not user and not password and not address and not port and not database):
raise ValueError("\n" + prompt2cn("数据库配置未填写", 28))
i_bind = bind
if not i_bind:
i_bind = f"{sql_name}://{user}:{password}@{address}:{port}/{database}"
update_info.json
+2 -1
View File
@@ -9,7 +9,8 @@
"configs/utils",
"poetry.lock",
"pyproject.toml",
"resources/font"
"resources/font",
"bot.py"
],
"add_file": [],
"delete_file": []
utils/image_utils.py
+1 -1
View File
@@ -1508,7 +1508,7 @@ async def text2image(
font=font,
)
await A.atext((left_padding, top_padding), text, font_color)
# A.show()
# A.show()
return A
utils/manager/resources_manager.py
+11 -8
View File
@@ -1,4 +1,6 @@
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 pathlib import Path
from ruamel.yaml import YAML
@@ -47,19 +49,21 @@ class ResourcesManager(StaticData):
if module in self._data.keys():
for x in self._data[module].keys():
move_file = Path(self._data[module][x])
if move_file.exists():
shutil.rmtree(move_file.absolute(), ignore_errors=True)
logger.info(f"已清除插件 {module} 资源路径：{self._data[module][x]}")
del self._data[module][x]
if move_file not in [IMAGE_PATH, DATA_PATH, RECORD_PATH, TEXT_PATH, FONT_PATH, LOG_PATH]:
if move_file.exists():
shutil.rmtree(move_file.absolute(), ignore_errors=True)
logger.info(f"已清除插件 {module} 资源路径：{self._data[module][x]}")
del self._data[module][x]
else:
if isinstance(source_file, Path):
source_file = str(source_file.absolute())
if source_file:
if module in self._data.keys() and source_file in self._data[module].keys():
move_file = Path(self._data[module][source_file])
if move_file.exists():
shutil.rmtree(move_file.absolute(), ignore_errors=True)
del self._data[module][source_file]
if move_file not in [IMAGE_PATH, DATA_PATH, RECORD_PATH, TEXT_PATH, FONT_PATH, LOG_PATH]:
if move_file.exists():
shutil.rmtree(move_file.absolute(), ignore_errors=True)
del self._data[module][source_file]
self.save()
def start_move(self):
@@ -97,7 +101,6 @@ class ResourcesManager(StaticData):
"""
添加临时清理文件夹
:param path: 路径
:param recursive: 是否将该目录下的所有目录也添加为临时文件夹
"""
if isinstance(path, str):
path = Path(path)
utils/text_utils.py
+12
View File
@@ -0,0 +1,12 @@
def prompt2cn(text: str, count: int, s: str = "#") -> str:
"""
格式化中文提示
:param text: 文本
:param count: 个数
:param s: #
"""
return s * count + "\n" + s * 6 + " " + text + " " + s * 6 + "\n" + s * count