Compare commits

..
32 Commits
Author SHA1 Message Date
HibiKier b3b09ed666 update v0.1.5.3 2022-05-09 21:42:24 +08:00
HibiKier 8029c00a02 Update README.md 2022-05-05 20:32:22 +08:00
HibiKier a2c0921013 Merge pull request #153 from NumberSir/main
修改bilibili_sub插件在windows平台下报错
2022-05-05 20:32:17 +08:00
Number_Sir 31cdabc229 修改bilibili_sub插件在windows平台下报错
改用bilireq库,同时补充部分bilireq暂无的方法如获取番剧、获取用户投稿视频
2022-05-04 17:02:56 +08:00
HibiKier 1059a0aae0 Update README.md 2022-05-03 15:40:19 +08:00
HibiKier e6576e7f40 Update README.md 2022-05-03 15:34:47 +08:00
HibiKier 5837144095 Update data_source.py 2022-05-03 15:31:34 +08:00
HibiKier d30a5c4207 Update data_source.py 2022-05-03 10:03:27 +08:00
HibiKier 7d169b0e5e Update README.md 2022-05-03 09:46:06 +08:00
HibiKier 1f54226673 update ShopRegister 2022-05-03 09:43:38 +08:00
HibiKier 901a90ff13 update v0.1.5.1 2022-05-03 02:03:06 +08:00
HibiKier 56573d1d34 update black_word 2022-05-01 15:53:52 +08:00
HibiKier b0ab78340a Update chat_message_handle.py 2022-04-26 15:16:26 +08:00
HibiKier 93539be492 update v0.1.5.0 2022-04-26 14:45:04 +08:00
HibiKier c551e21766 Update __version__ 2022-04-12 00:51:37 +08:00
HibiKier ed342af97d Update README.md 2022-04-12 00:51:25 +08:00
HibiKier 17b87a2cb1 Update README.md 2022-04-12 00:50:55 +08:00
HibiKier 69ca02834d Update __init__.py 2022-04-12 00:50:19 +08:00
HibiKier 6cb4099a2e Update README.md 2022-04-10 22:57:36 +08:00
HibiKier e75a4966b2 Update __init__.py 2022-04-10 22:57:05 +08:00
HibiKier d55cba49bd Update data_source.py 2022-04-10 22:31:07 +08:00
HibiKier 0215abac30 update v0.1.4.7 2022-04-10 22:19:50 +08:00
HibiKier 2dcac5f009 Update README.md 2022-04-09 18:09:22 +08:00
HibiKier 0d8c64f7d1 Merge pull request #144 from AkashiCoin/main
fix: 更新问题,戳一戳图片路径问题
2022-04-09 18:08:49 +08:00
AkashiCoin 55626645fb fix: 更新问题,戳一戳图片路径问题 2022-04-09 14:56:30 +08:00
HibiKier f7925fa259 update genshin.query_role 2022-04-08 01:09:29 +08:00
HibiKier c814ae08b1 Update README.md 2022-04-06 18:52:01 +08:00
HibiKier 24666368f7 Merge pull request #143 from yajiwa/main
update search_type
2022-04-06 18:50:59 +08:00
yajiwa a77e85b62d fix bug 2022-04-06 13:17:08 +08:00
yajiwa 6319c7fce1 fix 2022-04-05 22:38:43 +08:00
yajiwa a45b95d5d4 update search_type 2022-04-05 22:15:47 +08:00
HibiKier 188f6f0f99 Update README.md 2022-04-05 19:42:20 +08:00
57 changed files with 2515 additions and 623 deletions
-1
View File
@@ -142,6 +142,5 @@ test.py
server_ip.py
member_activity_handle.py
Yu-Gi-Oh/
black_word/
csgo/
fantasy_card/
+58
View File
@@ -32,6 +32,12 @@
## 普通帮助图片
![x](https://github.com/HibiKier/zhenxun_bot/blob/0.0.8.2/docs_image/3238573864-836268675-E2FFBB2AC143EAF4DDDF150438508721.png)
## Web UI
[zhenxun_bot_webui](https://github.com/HibiKier/zhenxun_bot_webui)
## 一键安装脚本
[zhenxun_bot-deploy](https://github.com/AkashiCoin/zhenxun_bot-deploy)
## 提供符合真寻标准的插件仓库
[AkashiCoin/nonebot_plugins_zhenxun_bot](https://github.com/AkashiCoin/nonebot_plugins_zhenxun_bot)
@@ -236,6 +242,58 @@ __Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
## 更新
### 2022/5/9
* 替换了疫情API
* 修复了私聊.ban/.unban出错
### 2022/5/5
* 修改bilibili_sub插件在windows平台下报错 [@pull/153](https://github.com/HibiKier/zhenxun_bot/pull/153)
### 2022/5/3 \[v0.1.5.2]
* 商品使用函数可以添加特定参数,例如:user_id, group_id, ShopParam等以及自己提供的参数
* 添加商品注册装饰器shop_register
* 修复商品函数kwargs无法获取参数值
### 2022/5/1
* 删除了`group_last_chat`插件(该功能可由`chat_history`替代
* 新增敏感词检测(全新反击系统,是时候重拳出击了
### 2022/4/26
* 修复了群白名单无法正确添加
* 优化了管理员帮助图片,背景图层将位于最下层
* 修复了树脂140时不断提醒(未测试
* 新增了消息记录的消息排行
* WebUI新增CPU,内存,磁盘监控
* WebUI新增资源文件夹统计可视化
### 2022/4/12
* 修复b了命令私聊出错
### 2022/4/10 \[v0.1.4.7]
* 新增消息记录模块
* 丰富处理请求操作提示
* web ui新增配置项修改
### 2022/4/9
* fix: 更新问题,戳一戳图片路径问题 [@pull/144](https://github.com/HibiKier/zhenxun_bot/pull/144)
### 2022/4/8
* 修复原神玩家查询
### 2022/4/6
* update search_type [@pull/143](https://github.com/HibiKier/zhenxun_bot/pull/143)
### 2022/4/5 \[v0.1.4.6]
* 修复web修改插件后帮助图片生成错误
+1 -1
View File
@@ -1 +1 @@
__version__: v0.1.4.6
__version__: v0.1.5.3
+5 -8
View File
@@ -5,7 +5,6 @@ from utils.utils import get_matchers
from utils.manager import group_manager
from nonebot.adapters.onebot.v11 import Bot
from nonebot import Driver
import asyncio
import nonebot
@@ -27,12 +26,10 @@ async def create_help_image():
"""
创建管理员帮助图片
"""
await asyncio.get_event_loop().run_in_executor(
None, _create_help_image
)
await _create_help_image()
def _create_help_image():
async def _create_help_image():
"""
创建管理员帮助图片
"""
@@ -85,9 +82,9 @@ def _create_help_image():
height = len(help_str.split("\n")) * 33
A = BuildImage(width, height, font_size=24)
_background = BuildImage(width, height, background=background)
A.text((150, 110), help_str)
A.paste(_background, alpha=True)
A.save(admin_help_image)
await A.apaste(_background, alpha=True)
await A.atext((150, 110), help_str)
await A.asave(admin_help_image)
logger.info(f'已成功加载 {len(_plugin_name_list)} 条管理员命令')
+46 -30
View File
@@ -1,15 +1,17 @@
from nonebot.adapters.onebot.v11 import GroupMessageEvent, PrivateMessageEvent, MessageEvent, Message, Bot
from nonebot.params import CommandArg, Command
from nonebot import on_command
from typing import Tuple
from configs.config import NICKNAME, Config
from models.ban_user import BanUser
from models.level_user import LevelUser
from typing import Tuple
from utils.utils import get_message_at, is_number
from configs.config import NICKNAME, Config
from nonebot import on_command
from nonebot.adapters.onebot.v11 import (Bot, GroupMessageEvent, Message,
MessageEvent, PrivateMessageEvent)
from nonebot.params import Command, CommandArg
from nonebot.permission import SUPERUSER
from .data_source import parse_ban_time, a_ban
from services.log import logger
from utils.utils import get_message_at, is_number
from .data_source import a_ban, parse_ban_time
__zx_plugin_name__ = "封禁Ban用户 [Admin]"
__plugin_usage__ = """
@@ -34,19 +36,19 @@ usage:
示例:b了 1234567
示例:.ban 12345567
""".strip()
__plugin_des__ = '你被逮捕了!丢进小黑屋!'
__plugin_cmd__ = ['.ban [at] ?[小时] ?[分钟]', '.unban [at]', 'b了 [at] [_superuser]']
__plugin_des__ = "你被逮捕了!丢进小黑屋!"
__plugin_cmd__ = [".ban [at] ?[小时] ?[分钟]", ".unban [at]", "b了 [at] [_superuser]"]
__plugin_version__ = 0.1
__plugin_author__ = 'HibiKier'
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"admin_level": Config.get_config("ban", "BAN_LEVEL"),
"cmd": ['.ban', '.unban', 'ban', 'unban']
"cmd": [".ban", ".unban", "ban", "unban"],
}
__plugin_configs__ = {
"BAN_LEVEL [LEVEL]": {
"value": 5,
"help": "ban/unban所需要的管理员权限等级",
"default_value": 5
"default_value": 5,
}
}
@@ -58,18 +60,23 @@ ban = on_command(
block=True,
)
super_ban = on_command('b了', permission=SUPERUSER, priority=5, block=True)
super_ban = on_command("b了", permission=SUPERUSER, priority=5, block=True)
@ban.handle()
async def _(bot: Bot, event: GroupMessageEvent, cmd: Tuple[str, ...] = Command(), arg: Message = CommandArg()):
async def _(
bot: Bot,
event: GroupMessageEvent,
cmd: Tuple[str, ...] = Command(),
arg: Message = CommandArg(),
):
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)
user_name = user_name['card'] or user_name['nickname']
user_name = user_name["card"] or user_name["nickname"]
msg = arg.extract_plain_text().strip()
time = parse_ban_time(msg)
if isinstance(time, str):
@@ -108,49 +115,58 @@ async def _(bot: Bot, event: GroupMessageEvent, cmd: Tuple[str, ...] = Command()
@ban.handle()
async def _(bot: Bot, event: PrivateMessageEvent, cmd: Tuple[str, ...] = Command(), arg: Message = CommandArg()):
async def _(
bot: Bot,
event: PrivateMessageEvent,
cmd: Tuple[str, ...] = Command(),
arg: Message = CommandArg(),
):
cmd = cmd[0]
msg = arg.extract_plain_text().strip()
if msg:
if str(event.user_id) in bot.config.superusers:
if is_number(arg.extract_plain_text().strip().split()[0]):
qq = int(msg[0])
msg = msg[1:]
msg_splt = msg.split()
if is_number(msg_splt[0]):
qq = int(msg_splt[0])
msg = msg_splt[1:]
if cmd in [".ban", "/ban"]:
time = parse_ban_time(msg)
time = parse_ban_time(" ".join(msg))
if isinstance(time, str):
await ban.finish(time)
result = await a_ban(qq, time, str(qq), event, 9)
else:
if await BanUser.unban(qq):
logger.info(
f"USER {event.user_id} 将 USER {qq} 解禁"
)
logger.info(f"USER {event.user_id} 将 USER {qq} 解禁")
result = f"已经把 {qq} 从黑名单中删除了!"
else:
result = f"{qq} 不在黑名单!"
await ban.send(result)
else:
await ban.finish('qq号必须是数字!\n格式:.ban [qq] [hour]? [minute]?', at_sender=True)
await ban.finish(
"qq号必须是数字!\n格式:.ban [qq] [hour]? [minute]?", at_sender=True
)
@super_ban.handle()
async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
user_name = ""
if isinstance(event, GroupMessageEvent):
qq = get_message_at(event.json())
if qq:
qq = qq[0]
user = await bot.get_group_member_info(group_id=event.group_id, user_id=qq)
user_name = user["card"] or user["nickname"]
else:
qq = arg.extract_plain_text().strip()
if not is_number(qq):
await super_ban.finish("对象qq必须为纯数字...")
qq = [qq]
qq = int(qq)
user_name = qq
if qq:
qq = qq[0]
user = await bot.get_group_member_info(group_id=event.group_id, user_id=qq)
user_name = user['card'] or user['nickname']
if not await BanUser.ban(qq, 10, 99999999):
await BanUser.unban(qq)
await BanUser.ban(qq, 10, 99999999)
await ban.send(f"已将 {user_name} 拉入黑名单!")
else:
await super_ban.send('需要添加被super ban的对象,可以使用at或者指定qq..')
await super_ban.send("需要添加被super ban的对象,可以使用at或者指定qq..")
+5 -12
View File
@@ -25,7 +25,9 @@ def parse_ban_time(msg: str) -> Union[int, str]:
return int(msg[0]) * 60 * 60 + int(msg[1]) * 60
async def a_ban(qq: int, time: int, user_name: str, event: MessageEvent, ban_level: int = None) -> str:
async def a_ban(
qq: int, time: int, user_name: str, event: MessageEvent, ban_level: int = None
) -> str:
"""
ban
:param qq: qq
@@ -38,7 +40,8 @@ async def a_ban(qq: int, time: int, user_name: str, event: MessageEvent, ban_lev
ban_level = await LevelUser.get_user_level(event.user_id, event.group_id)
if await BanUser.ban(qq, ban_level, time):
logger.info(
f"USER {event.user_id} GROUP {event.group_id} 将 USER {qq} 封禁 时长 {time / 60} 分钟"
f"USER {event.user_id} GROUP"
f" {event.group_id if isinstance(event, GroupMessageEvent) else ''} 将 USER {qq} 封禁 时长 {time / 60} 分钟"
)
result = f"已经将 {user_name} 加入{NICKNAME}的黑名单了!"
if time != -1:
@@ -57,13 +60,3 @@ async def a_ban(qq: int, time: int, user_name: str, event: MessageEvent, ban_lev
time += " 分钟"
result = f"{user_name} 已在黑名单!预计 {time}后解封"
return result
+3
View File
@@ -0,0 +1,3 @@
import nonebot
nonebot.load_plugins("basic_plugins/chat_history")
+6
View File
@@ -0,0 +1,6 @@
from nonebot.adapters.onebot.v11 import Event, MessageEvent
from configs.config import Config
def rule(event: Event) -> bool:
return Config.get_config("chat_history", "FLAG") and isinstance(event, MessageEvent)
@@ -0,0 +1,38 @@
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 ._rule import rule
__zx_plugin_name__ = "消息存储 [Hidden]"
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
Config.add_plugin_config(
"chat_history", "FLAG", True, help_="是否开启消息自从存储", name="消息存储", default_value=True
)
chat_history = on_message(rule=rule, priority=1, block=False)
@chat_history.handle()
async def _(event: MessageEvent):
if isinstance(event, GroupMessageEvent):
await ChatHistory.add_chat_msg(
event.user_id, event.group_id, str(event.get_message())
)
else:
await ChatHistory.add_chat_msg(event.user_id, None, str(event.get_message()))
# @test.handle()
# async def _(event: MessageEvent):
# print(await ChatHistory.get_user_msg(event.user_id, "private"))
# print(await ChatHistory.get_user_msg_count(event.user_id, "private"))
# print(await ChatHistory.get_user_msg(event.user_id, "group"))
# print(await ChatHistory.get_user_msg_count(event.user_id, "group"))
# print(await ChatHistory.get_group_msg(event.group_id))
# print(await ChatHistory.get_group_msg_count(event.group_id))
@@ -0,0 +1,108 @@
from datetime import datetime, timedelta
import pytz
from models.chat_history import ChatHistory
from models.group_member_info import GroupInfoUser
from nonebot import on_regex
from nonebot.adapters.onebot.v11 import GroupMessageEvent
from nonebot.params import RegexGroup
from utils.image_utils import BuildImage, text2image
from utils.utils import is_number
from utils.message_builder import image
from typing import Tuple, Any
__zx_plugin_name__ = "消息统计"
__plugin_usage__ = """
usage:
发言记录统计
regex:(周|月)?消息排行(des|DES)?(n=[0-9]{1,2})?
指令:
消息统计?(des)?(n=?)
周消息统计?(des)?(n=?)
月消息统计?(des)?(n=?)
示例:
消息统计
消息统计des
消息统计DESn=15
消息统计n=15
""".strip()
__plugin_des__ = "发言消息排行"
__plugin_cmd__ = [
"消息统计",
"周消息统计",
"月消息统计"
]
__plugin_type__ = ("数据统计", 1)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"cmd": ["消息统计"],
}
msg_handler = on_regex(r"^(周|月)?消息统计(des|DES)?(n=[0-9]{1,2})?$", priority=5, block=True)
@msg_handler.handle()
async def _(event: GroupMessageEvent, reg_group: Tuple[Any, ...] = RegexGroup()):
gid = event.group_id
date_scope = None
date, order, num = reg_group
num = num.split("=")[-1] if num else 10
if num and is_number(num) and 10 < int(num) < 50:
num = int(num)
if date in ["周"]:
date_scope = (datetime.now() - timedelta(days=7), datetime.now())
elif date in ["月"]:
date_scope = (datetime.now() - timedelta(days=30), datetime.now())
if rank_data := await ChatHistory.get_group_msg_rank(
gid, num, order or "DESC", date_scope
):
name = "昵称:\n\n"
num_str = "发言次数:\n\n"
idx = 1
for uid, num in rank_data:
try:
user_name = (await GroupInfoUser.get_member_info(uid, gid)).user_name
except AttributeError:
user_name = uid
name += f"\t{idx}.{user_name} \n\n"
num_str += f"\t{num}\n\n"
idx += 1
name_img = await text2image(name.strip(), padding=10, color="#f9f6f2")
num_img = await text2image(num_str.strip(), padding=10, color="#f9f6f2")
if not date_scope:
if date_scope := await ChatHistory.get_group_first_msg_datetime(gid):
date_scope = date_scope.astimezone(
pytz.timezone("Asia/Shanghai")
).replace(microsecond=0)
else:
date_scope = datetime.now().replace(microsecond=0)
date_str = f"日期:{date_scope} - 至今"
else:
date_str = f"日期:{date_scope[0].replace(microsecond=0)} - {date_scope[1].replace(microsecond=0)}"
date_w = BuildImage(0, 0, font_size=15).getsize(date_str)[0]
img_w = date_w if date_w > name_img.w + num_img.w else name_img.w + num_img.w
A = BuildImage(
img_w + 15,
num_img.h + 30,
color="#f9f6f2",
font="CJGaoDeGuo.otf",
font_size=15,
)
await A.atext((10, 10), date_str)
await A.apaste(name_img, (0, 30))
await A.apaste(num_img, (name_img.w, 30))
await msg_handler.send(image(b64=A.pic2bs4()))
# @test.handle()
# async def _(event: MessageEvent):
# print(await ChatHistory.get_user_msg(event.user_id, "private"))
# print(await ChatHistory.get_user_msg_count(event.user_id, "private"))
# print(await ChatHistory.get_user_msg(event.user_id, "group"))
# print(await ChatHistory.get_user_msg_count(event.user_id, "group"))
# print(await ChatHistory.get_group_msg(event.group_id))
# print(await ChatHistory.get_group_msg_count(event.group_id))
+2 -2
View File
@@ -277,7 +277,7 @@ def _create_help_img(
0,
plain_text=msg,
font_size=24,
font="yuanshen.ttf",
font="HYWenHei-85W.ttf",
)
B.paste(text, (w, h), True)
h += 50
@@ -289,7 +289,7 @@ def _create_help_img(
0,
plain_text="注: 红字代表功能被群管理员禁用,红线代表功能正在维护",
font_size=24,
font="yuanshen.ttf",
font="HYWenHei-85W.ttf",
font_color=(231, 74, 57)
),
(300, 10),
+1 -1
View File
@@ -10,7 +10,7 @@ async def handle_api_call(bot: Bot, api: str, data: Dict[str, Any]):
r = None
if (
(
(api == "send_msg" and data["message_type"] == "group")
(api == "send_msg" and data.get("message_type") == "group")
or api == "send_group_msg"
)
and (
+52 -34
View File
@@ -7,6 +7,7 @@ from typing import Optional, Union
from configs.config import Config
from nonebot import Driver
from nonebot.plugin import require
from utils.decorator.shop import shop_register
import nonebot
import time
@@ -20,24 +21,26 @@ async def init_default_shop_goods():
"""
导入内置的三个商品
"""
async def sign_card(**kwargs):
user_id = kwargs['user_id']
group_id = kwargs['group_id']
prob = kwargs["prob"]
@shop_register(
name=("好感度双倍加持卡Ⅰ", "好感度双倍加持卡Ⅱ", "好感度双倍加持卡Ⅲ"),
price=(30, 150, 250),
des=(
"下次签到双倍好感度概率 + 10%(谁才是真命天子?)(同类商品将覆盖)",
"下次签到双倍好感度概率 + 20%(平平庸庸)(同类商品将覆盖)",
"下次签到双倍好感度概率 + 30%(金币才是真命天子!)(同类商品将覆盖)",
),
load_status=Config.get_config("shop", "IMPORT_DEFAULT_SHOP_GOODS"),
** {"好感度双倍加持卡Ⅰ_prob": 0.1, "好感度双倍加持卡Ⅱ_prob": 0.2, "好感度双倍加持卡Ⅲ_prob": 0.3},
)
async def sign_card(user_id: int, group_id: int, prob: float):
user = await SignGroupUser.ensure(user_id, group_id)
await user.update(add_probability=prob).apply()
if Config.get_config("shop", "IMPORT_DEFAULT_SHOP_GOODS"):
await register_goods(
"好感度双倍加持卡Ⅰ", 30, "下次签到双倍好感度概率 + 10%(谁才是真命天子?)(同类商品将覆盖)"
)
use.register_use("好感度双倍加持卡Ⅰ", sign_card, **{"prob": 0.1})
await register_goods("好感度双倍加持卡Ⅱ", 150, "下次签到双倍好感度概率 + 20%(平平庸庸)(同类商品将覆盖)")
use.register_use("好感度双倍加持卡Ⅱ", sign_card, **{"prob": 0.2})
await register_goods(
"好感度双倍加持卡Ⅲ", 250, "下次签到双倍好感度概率 + 30%(金币才是真命天子!)(同类商品将覆盖)"
)
use.register_use("好感度双倍加持卡Ⅲ", sign_card, **{"prob": 0.3})
@driver.on_bot_connect
async def _():
await shop_register.load_register()
# 创建商店界面
@@ -59,9 +62,7 @@ async def create_shop_help() -> str:
A = BuildImage(1000, h, color="#f9f6f2")
current_h = 0
for goods in _list:
bk = BuildImage(
700, 80, font_size=15, color="#f9f6f2", font="CJGaoDeGuo.otf"
)
bk = BuildImage(700, 80, font_size=15, color="#f9f6f2", font="CJGaoDeGuo.otf")
goods_image = BuildImage(
600, 80, font_size=20, color="#a29ad6", font="CJGaoDeGuo.otf"
)
@@ -92,10 +93,18 @@ async def create_shop_help() -> str:
await bk.apaste(goods_image, alpha=True)
# 添加限时图标和时间
if goods.goods_limit_time > 0:
_limit_time_logo = BuildImage(40, 40, background=f"{IMAGE_PATH}/other/time.png")
_limit_time_logo = BuildImage(
40, 40, background=f"{IMAGE_PATH}/other/time.png"
)
await bk.apaste(_limit_time_logo, (600, 0), True)
await bk.apaste(BuildImage(0, 0, plain_text="限时!", font_size=23, font="CJGaoDeGuo.otf"), (640, 10), True)
limit_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(goods.goods_limit_time)).split()
await bk.apaste(
BuildImage(0, 0, plain_text="限时!", font_size=23, font="CJGaoDeGuo.otf"),
(640, 10),
True,
)
limit_time = time.strftime(
"%Y-%m-%d %H:%M", time.localtime(goods.goods_limit_time)
).split()
y_m_d = limit_time[0]
_h_m = limit_time[1].split(":")
h_m = _h_m[0] + "时 " + _h_m[1] + "分"
@@ -125,12 +134,12 @@ async def create_shop_help() -> str:
async def register_goods(
name: str,
price: int,
des: str,
discount: Optional[float] = 1,
limit_time: Optional[int] = 0,
**kwargs,
name: str,
price: int,
des: str,
discount: Optional[float] = 1,
limit_time: Optional[int] = 0,
**kwargs,
):
"""
添加商品
@@ -151,12 +160,17 @@ async def register_goods(
des = kwargs.get("des")
discount = kwargs.get("discount")
limit_time = kwargs.get("time_limit")
limit_time = float(limit_time) if limit_time else limit_time
discount = discount if discount is None else 1
limit_time = int(time.time() + limit_time * 60 * 60) if limit_time is not None and limit_time != 0 else 0
return await GoodsInfo.add_goods(
name, int(price), des, float(discount), limit_time
)
if await GoodsInfo.get_goods_info(name):
limit_time = float(limit_time) if limit_time else limit_time
discount = discount if discount is None else 1
limit_time = (
int(time.time() + limit_time * 60 * 60)
if limit_time is not None and limit_time != 0
else 0
)
return await GoodsInfo.add_goods(
name, int(price), des, float(discount), limit_time
)
# 删除商品
@@ -230,7 +244,11 @@ async def update_goods(**kwargs) -> "str, str, int":
int(price),
des,
float(discount),
int(time.time() + limit_time * 60 * 60 if limit_time != 0 and new_time else 0),
int(
time.time() + limit_time * 60 * 60
if limit_time != 0 and new_time
else 0
),
),
name,
tmp[:-1],
+80 -31
View File
@@ -1,7 +1,11 @@
from nonebot.adapters.onebot.v11 import GroupMessageEvent, MessageSegment
from services.log import logger
from nonebot.adapters.onebot.v11 import Bot
from pydantic import create_model
from utils.models import ShopParam
from typing import Optional, Union
from types import MappingProxyType
import inspect
import asyncio
@@ -30,24 +34,52 @@ class GoodsUseFuncManager:
:param goods_name: 商品名称
"""
if self.exists(goods_name):
return self._data[goods_name]["kwargs"]["_max_num_limit"]
return self._data[goods_name]["kwargs"]["max_num_limit"]
return 1
async def use(self, **kwargs) -> Optional[Union[str, MessageSegment]]:
async def use(
self, param: ShopParam, **kwargs
) -> Optional[Union[str, MessageSegment]]:
"""
使用道具
:param param: BaseModel
:param kwargs: kwargs
"""
goods_name = kwargs.get("goods_name")
def parse_args(args_: MappingProxyType):
param_list_ = []
_bot = param.bot
param.bot = None
param_json = param.dict()
param_json["bot"] = _bot
for par in args_.keys():
if par in ["shop_param"]:
param_list_.append(param)
elif par not in ["args", "kwargs"]:
param_list_.append(param_json.get(par))
if kwargs.get(par) is not None:
del kwargs[par]
return param_list_
goods_name = param.goods_name
if self.exists(goods_name):
if asyncio.iscoroutinefunction(self._data[goods_name]["func"]):
return await self._data[goods_name]["func"](
**kwargs,
)
args = inspect.signature(self._data[goods_name]["func"]).parameters
if args and list(args.keys())[0] != "kwargs":
if asyncio.iscoroutinefunction(self._data[goods_name]["func"]):
return await self._data[goods_name]["func"](
*parse_args(args)
)
else:
return self._data[goods_name]["func"](
*parse_args(args)
)
else:
return self._data[goods_name]["func"](
**kwargs,
)
if asyncio.iscoroutinefunction(self._data[goods_name]["func"]):
return await self._data[goods_name]["func"](
**kwargs,
)
else:
return self._data[goods_name]["func"](
**kwargs,
)
def check_send_success_message(self, goods_name: str) -> bool:
"""
@@ -67,6 +99,18 @@ class GoodsUseFuncManager:
return self._data[goods_name]["kwargs"]
return {}
def init_model(self, goods_name: str, bot: Bot, event: GroupMessageEvent, num: int):
return self._data[goods_name]["model"](
**{
"goods_name": goods_name,
"bot": bot,
"event": event,
"user_id": event.user_id,
"group_id": event.group_id,
"num": num,
}
)
func_manager = GoodsUseFuncManager()
@@ -83,22 +127,23 @@ async def effect(
:return: 使用是否成功
"""
# 优先使用注册的商品插件
try:
if func_manager.exists(goods_name):
_kwargs = func_manager.get_kwargs(goods_name)
return await func_manager.use(
**{
**_kwargs,
"_bot": bot,
"event": event,
"group_id": event.group_id,
"user_id": event.user_id,
"num": num,
"goods_name": goods_name,
}
)
except Exception as e:
logger.error(f"use 商品生效函数effect 发生错误 {type(e)}:{e}")
# try:
if func_manager.exists(goods_name):
_kwargs = func_manager.get_kwargs(goods_name)
return await func_manager.use(
func_manager.init_model(goods_name, bot, event, num),
**{
**_kwargs,
"_bot": bot,
"event": event,
"group_id": event.group_id,
"user_id": event.user_id,
"num": num,
"goods_name": goods_name,
},
)
# except Exception as e:
# logger.error(f"use 商品生效函数effect 发生错误 {type(e)}:{e}")
return None
@@ -112,10 +157,14 @@ def register_use(goods_name: str, func, **kwargs):
if func_manager.exists(goods_name):
raise ValueError("该商品使用函数已被注册!")
# 发送使用成功信息
if kwargs.get("send_success_msg") is None:
kwargs["send_success_msg"] = True
kwargs["_max_num_limit"] = (
kwargs.get("_max_num_limit") if kwargs.get("_max_num_limit") else 1
kwargs["send_success_msg"] = kwargs.get("send_success_msg", True)
kwargs["max_num_limit"] = kwargs.get("max_num_limit", 1)
func_manager.register_use(
goods_name,
**{
"func": func,
"model": create_model(f"{goods_name}_model", __base__=ShopParam, **kwargs),
"kwargs": kwargs,
},
)
func_manager.register_use(goods_name, **{"func": func, "kwargs": kwargs})
logger.info(f"register_use 成功注册商品:{goods_name} 的使用函数")
+20 -15
View File
@@ -82,15 +82,16 @@ async def _(bot: Bot, cmd: Tuple[str, ...] = Command(), arg: Message = CommandAr
if is_number(id_):
id_ = int(id_)
if cmd[:2] == "同意":
if await requests_manager.approve(bot, id_, "private"):
await friend_handle.send("同意好友请求成功..")
else:
await friend_handle.send("同意好友请求失败,可能是未找到此id的请求..")
flag = await requests_manager.approve(bot, id_, "private")
else:
if await requests_manager.refused(bot, id_, "private"):
await friend_handle.send("拒绝好友请求成功..")
else:
await friend_handle.send("拒绝好友请求失败,可能是未找到此id的请求..")
flag = await requests_manager.refused(bot, id_, "private")
if flag == 1:
await friend_handle.send(f"{cmd[:2]}好友请求失败,该请求已失效..")
requests_manager.delete_request(id_, "private")
elif flag == 2:
await friend_handle.send(f"{cmd[:2]}好友请求失败,未找到此id的请求..")
else:
await friend_handle.send(f"{cmd[:2]}好友请求成功!")
else:
await friend_handle.send("id必须为纯数字!")
@@ -99,12 +100,12 @@ async def _(bot: Bot, cmd: Tuple[str, ...] = Command(), arg: Message = CommandAr
async def _(bot: Bot, cmd: Tuple[str, ...] = Command(), arg: Message = CommandArg()):
cmd = cmd[0]
id_ = arg.extract_plain_text().strip()
flag = None
if is_number(id_):
id_ = int(id_)
if cmd[:2] == "同意":
rid = requests_manager.get_group_id(id_)
if rid:
await friend_handle.send("同意群聊请求成功..")
if await GroupInfo.get_group_info(rid):
await GroupInfo.set_group_flag(rid, 1)
else:
@@ -116,14 +117,18 @@ async def _(bot: Bot, cmd: Tuple[str, ...] = Command(), arg: Message = CommandAr
group_info["member_count"],
1
)
await requests_manager.approve(bot, id_, "group")
flag = await requests_manager.approve(bot, id_, "group")
else:
await friend_handle.send("同意群聊请求失败,可能是未找到此id的请求..")
await friend_handle.send("同意群聊请求失败,未找到此id的请求..")
else:
if await requests_manager.refused(bot, id_, "group"):
await friend_handle.send("拒绝群聊请求成功..")
else:
await friend_handle.send("拒绝群聊请求失败,可能是未找到此id的请求..")
flag = await requests_manager.refused(bot, id_, "group")
if flag == 1:
await friend_handle.send(f"{cmd[:2]}群聊请求失败,该请求已失效..")
requests_manager.delete_request(id_, "group")
elif flag == 2:
await friend_handle.send(f"{cmd[:2]}群聊请求失败,未找到此id的请求..")
else:
await friend_handle.send(f"{cmd[:2]}群聊请求成功!")
else:
await friend_handle.send("id必须为纯数字!")
+1 -1
View File
@@ -136,7 +136,7 @@ async def _():
@manager_group_whitelist.handle()
async def _(bot: Bot, cmd: Tuple[str, ...] = Command(), arg: Message = CommandArg()):
cmd = cmd[0]
msg = arg.extract_plain_text().strip()
msg = arg.extract_plain_text().strip().split()
all_group = [
g["group_id"] for g in await bot.get_group_list()
]
+19 -3
View File
@@ -11,14 +11,19 @@ class ConfigsManager:
def __init__(self, file: Path):
self._data: dict = {}
self._simple_data: dict = {}
self._admin_level_data = []
self._simple_file = Path() / "configs" / "config.yaml"
if file:
file.parent.mkdir(exist_ok=True, parents=True)
self.file = file
_yaml = YAML()
if file.exists():
_yaml = YAML()
with open(file, "r", encoding="utf8") as f:
self._data = _yaml.load(f)
if self._simple_file.exists():
with open(self._simple_file, "r", encoding="utf8") as f:
self._simple_data = _yaml.load(f)
def add_plugin_config(
self,
@@ -71,6 +76,7 @@ class ConfigsManager:
"""
if module in self._data.keys():
del self._data[module]
self.save()
def set_config(self, module: str, key: str, value: str):
"""
@@ -80,8 +86,10 @@ class ConfigsManager:
:param value: 值
"""
if module in self._data.keys():
if self._data[module].get(key) is not None:
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()
def set_help(self, module: str, key: str, help_: str):
"""
@@ -93,6 +101,7 @@ class ConfigsManager:
if module in self._data.keys():
if self._data[module].get(key) is not None:
self._data[module][key]["help"] = help_
self.save()
def set_default_value(self, module: str, key: str, value: str):
"""
@@ -104,6 +113,7 @@ class ConfigsManager:
if module in self._data.keys():
if self._data[module].get(key) is not None:
self._data[module][key]["default_value"] = value
self.save()
def get_config(self, module: str, key: str, default: Optional[Any] = None) -> Optional[Any]:
"""
@@ -142,11 +152,17 @@ class ConfigsManager:
if key in self._data.keys():
return self._data[key]
def save(self, path: Union[str, Path] = None):
def save(self, path: Union[str, Path] = None, save_simple_data: bool = False):
"""
保存数据
:param path: 路径
:param save_simple_data: 同时保存至config.yaml
"""
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
)
path = path if path else self.file
with open(path, "w", encoding="utf8") as f:
yaml.dump(
+214
View File
@@ -0,0 +1,214 @@
from datetime import datetime, timedelta
from typing import List, Literal, Optional, Tuple, Union
from services.db_context import db
class ChatHistory(db.Model):
__tablename__ = "chat_history"
id = db.Column(db.Integer(), primary_key=True)
user_qq = db.Column(db.BigInteger(), nullable=False)
group_id = db.Column(db.BigInteger())
text = db.Column(db.Text())
create_time = db.Column(db.DateTime(timezone=True), nullable=False)
@classmethod
async def add_chat_msg(cls, user_qq: int, group_id: Optional[int], text: str):
await cls.create(
user_qq=user_qq, group_id=group_id, text=text, create_time=datetime.now()
)
@classmethod
async def get_user_msg(
cls,
uid: int,
msg_type: Optional[Literal["private", "group"]],
days: Optional[int] = None,
) -> List["ChatHistory"]:
"""
说明:
获取用户消息
参数:
:param uid: 用户qq
:param msg_type: 消息类型,私聊或群聊
:param days: 限制日期
"""
return await cls._get_msg(uid, None, "user", msg_type, days).gino.all()
@classmethod
async def get_group_user_msg(
cls,
uid: int,
gid: int,
limit: int = 10,
date_scope: Tuple[datetime, datetime] = None,
) -> List["ChatHistory"]:
"""
说明:
获取群聊指定用户聊天记录
参数:
:param uid: qq
:param gid: 群号
:param limit: 获取数量
:param date_scope: 日期范围,默认None为全搜索
"""
return (
await cls._get_msg(uid, gid, "group", days=date_scope)
.limit(limit)
.gino.all()
)
@classmethod
async def get_group_user_msg_count(cls, uid: int, gid: int) -> Optional[int]:
"""
说明:
查询群聊指定用户的聊天记录数量
参数:
:param uid: qq
:param gid: 群号
"""
if x := await db.first(
db.text(
f"SELECT COUNT(id) as sum FROM public.chat_history WHERE user_qq = {uid} AND group_id = {gid}"
)
):
return x[0]
return None
@classmethod
async def get_group_msg_rank(
cls,
gid: int,
limit: int = 10,
order: str = "DESC",
date_scope: Optional[Tuple[datetime, datetime]] = None,
) -> Optional[Tuple[int, int]]:
"""
说明:
获取排行数据
参数:
:param gid: 群号
:param limit: 获取数量
:param order: 排序类型,desc,des
:param date_scope: 日期范围
"""
sql = f"SELECT user_qq, COUNT(id) as sum FROM public.chat_history WHERE group_id = {gid} "
if date_scope:
sql += f"AND create_time BETWEEN '{date_scope[0]}' AND '{date_scope[1]}' "
sql += f"GROUP BY user_qq ORDER BY sum {order if order and order.upper() != 'DES' else ''} LIMIT {limit}"
print(sql)
return await db.all(db.text(sql))
@classmethod
async def get_group_first_msg_datetime(cls, gid: int) -> Optional[datetime]:
"""
说明:
获取群第一条记录消息时间
参数:
:param gid:
"""
if (
msg := await cls.query.where(cls.group_id == gid)
.order_by(cls.create_time)
.gino.first()
):
return msg.create_time
return None
@classmethod
async def get_user_msg_count(
cls,
uid: int,
msg_type: Optional[Literal["private", "group"]],
days: Optional[int] = None,
) -> int:
"""
说明:
获取用户消息数量
参数:
:param uid: 用户qq
:param msg_type: 消息类型,私聊或群聊
:param days: 限制日期
"""
return (
await cls._get_msg(uid, None, "user", msg_type, days, True).gino.first()
)[0]
@classmethod
async def get_group_msg(
cls,
gid: int,
days: Optional[int] = None,
) -> List["ChatHistory"]:
"""
说明:
获取群聊消息
参数:
:param gid: 用户qq
:param days: 限制日期
"""
return await cls._get_msg(None, gid, "group", None, days).gino.all()
@classmethod
async def get_group_msg_count(
cls,
gid: int,
days: Optional[int] = None,
) -> List["ChatHistory"]:
"""
说明:
获取群聊消息数量
参数:
:param gid: 用户qq
:param days: 限制日期
"""
return (await cls._get_msg(None, gid, "group", None, days, True).gino.first())[
0
]
@classmethod
def _get_msg(
cls,
uid: Optional[int],
gid: Optional[int],
type_: Literal["user", "group"],
msg_type: Optional[Literal["private", "group"]] = None,
days: Optional[Union[int, Tuple[datetime, datetime]]] = None,
is_select_count: bool = False,
):
"""
说明:
获取消息查询query
参数:
:param uid: 用户qq
:param gid: 群号
:param type_: 类型,私聊或群聊
:param msg_type: 消息类型,用户或群聊
:param days: 限制日期
"""
if is_select_count:
setattr(ChatHistory, "count", db.func.count(cls.id).label("count"))
query = cls.select("count")
else:
query = cls.query
if type_ == "user":
query = query.where(cls.user_qq == uid)
if msg_type == "private":
query = query.where(cls.group_id == None)
elif msg_type == "group":
query = query.where(cls.group_id != None)
else:
query = query.where(cls.group_id == gid)
if uid:
query = query.where(cls.user_qq == uid)
if days:
if isinstance(days, int):
query = query.where(
cls.create_time >= datetime.now() - timedelta(days=days)
)
elif isinstance(days, tuple):
query = query.where(cls.create_time >= days[0]).where(
cls.create_time <= days[1]
)
return query
+1
View File
@@ -76,6 +76,7 @@ async def _(bot: Bot, event: MessageEvent):
f"问题:{msg} ---- 回答:{result}"
)
if result:
result = str(result)
for t in Config.get_config("ai", "TEXT_FILTER"):
result = result.replace(t, "*")
await ai.finish(Message(result))
+33 -29
View File
@@ -126,35 +126,39 @@ async def xie_ai(text: str) -> str:
"""
res = await AsyncHttpx.get(f"http://api.qingyunke.com/api.php?key=free&appid=0&msg={text}")
content = ""
data = json.loads(res.text)
if data["result"] == 0:
content = data["content"]
if "菲菲" in content:
content = content.replace("菲菲", NICKNAME)
if "艳儿" in content:
content = content.replace("艳儿", NICKNAME)
if "公众号" in content:
content = ""
if "{br}" in content:
content = content.replace("{br}", "\n")
if "提示" in content:
content = content[: content.find("提示")]
if "淘宝" in content or "taobao.com" in content:
return ""
while True:
r = re.search("{face:(.*)}", content)
if r:
id_ = r.group(1)
content = content.replace(
"{" + f"face:{id_}" + "}", str(face(int(id_)))
)
else:
break
return (
content
if not content and not Config.get_config("ai", "ALAPI_AI_CHECK")
else await check_text(content)
)
try:
data = json.loads(res.text)
if data["result"] == 0:
content = data["content"]
if "菲菲" in content:
content = content.replace("菲菲", NICKNAME)
if "艳儿" in content:
content = content.replace("艳儿", NICKNAME)
if "公众号" in content:
content = ""
if "{br}" in content:
content = content.replace("{br}", "\n")
if "提示" in content:
content = content[: content.find("提示")]
if "淘宝" in content or "taobao.com" in content:
return ""
while True:
r = re.search("{face:(.*)}", content)
if r:
id_ = r.group(1)
content = content.replace(
"{" + f"face:{id_}" + "}", str(face(int(id_)))
)
else:
break
return (
content
if not content and not Config.get_config("ai", "ALAPI_AI_CHECK")
else await check_text(content)
)
except Exception as e:
logger.error(f"Ai xie_ai 发生错误 {type(e)}:{e}")
return ""
def hello() -> str:
+1 -1
View File
@@ -38,7 +38,7 @@ usage:
__plugin_des__ = "非常便利的B站订阅通知"
__plugin_cmd__ = ["添加订阅 [主播/UP/番剧] [id/链接/番名]", "删除订阅 [id]", "查看订阅"]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_author__ = "HibiKier & NumberSir"
__plugin_settings__ = {
"level": 5,
"default_status": True,
+40 -28
View File
@@ -1,12 +1,18 @@
from bilibili_api.exceptions.ResponseCodeException import ResponseCodeException
# from bilibili_api.exceptions.ResponseCodeException import ResponseCodeException
from bilireq.exceptions import ResponseCodeError
from utils.manager import resources_manager
from asyncio.exceptions import TimeoutError
from .model import BilibiliSub
from bilibili_api.live import LiveRoom
from bilibili_api import bangumi
# from bilibili_api.live import LiveRoom
from bilireq.live import get_room_info_by_id
# from bilibili_api import bangumi
from .utils import get_meta
from utils.message_builder import image
from bilibili_api.user import User
from bilibili_api import user
# from bilibili_api import user
# from bilibili_api.user import User
from bilireq.user import get_user_info
from bilireq import dynamic
from .utils import get_videos
from typing import Optional
from configs.path_config import IMAGE_PATH
from datetime import datetime
@@ -36,9 +42,9 @@ async def add_live_sub(live_id: int, sub_user: str) -> str:
try:
async with db.transaction():
try:
live = LiveRoom(live_id)
live_info = (await live.get_room_info())["room_info"]
except ResponseCodeException:
"""bilibili_api.live库的LiveRoom类中get_room_info改为bilireq.live库的get_room_info_by_id方法"""
live_info = await get_room_info_by_id(live_id)
except ResponseCodeError:
return f"未找到房间号Id:{live_id} 的信息,请检查Id是否正确"
uid = live_info["uid"]
room_id = live_info["room_id"]
@@ -78,16 +84,18 @@ async def add_up_sub(uid: int, sub_user: str) -> str:
try:
async with db.transaction():
try:
u = user.User(uid)
user_info = await u.get_user_info()
except ResponseCodeException:
"""bilibili_api.user库中User类的get_user_info改为bilireq.user库的get_user_info方法"""
user_info = await get_user_info(uid)
except ResponseCodeError:
return f"未找到UpId:{uid} 的信息,请检查Id是否正确"
uname = user_info["name"]
dynamic_info = await u.get_dynamics(0)
"""bilibili_api.user库中User类的get_dynamics改为bilireq.dynamic库的get_user_dynamics方法"""
dynamic_info = await dynamic.get_user_dynamics(uid)
dynamic_upload_time = 0
if dynamic_info.get("cards"):
dynamic_upload_time = dynamic_info["cards"][0]["desc"]["timestamp"]
video_info = await u.get_videos()
"""bilibili_api.user库中User类的get_videos改为bilireq.user库的get_videos方法"""
video_info = await get_videos(uid)
latest_video_created = 0
if video_info["list"].get("vlist"):
latest_video_created = video_info["list"]["vlist"][0]["created"]
@@ -117,8 +125,9 @@ async def add_season_sub(media_id: int, sub_user: str) -> str:
try:
async with db.transaction():
try:
season_info = await bangumi.get_meta(media_id)
except ResponseCodeException:
"""bilibili_api.bangumi库中get_meta改为bilireq.bangumi库的get_meta方法"""
season_info = await get_meta(media_id)
except ResponseCodeError:
return f"未找到media_id:{media_id} 的信息,请检查Id是否正确"
season_id = season_info["media"]["season_id"]
season_current_episode = season_info["media"]["new_ep"]["index"]
@@ -200,7 +209,7 @@ async def get_sub_status(id_: int, sub_type: str) -> Optional[str]:
return await _get_up_status(id_)
elif sub_type == "season":
return await _get_season_status(id_)
except ResponseCodeException:
except ResponseCodeError:
return f"Id:{id_} 获取信息失败...请检查订阅Id是否存在或稍后再试..."
# except Exception as e:
# logger.error(f"获取订阅状态发生预料之外的错误 id_:{id_} {type(e)}:{e}")
@@ -212,12 +221,12 @@ async def _get_live_status(id_: int) -> Optional[str]:
获取直播订阅状态
:param id_: 直播间 id
"""
live = LiveRoom(id_)
live_info = (await live.get_room_info())["room_info"]
"""bilibili_api.live库的LiveRoom类中get_room_info改为bilireq.live库的get_room_info_by_id方法"""
live_info = await get_room_info_by_id(id_)
title = live_info["title"]
room_id = live_info["room_id"]
live_status = live_info["live_status"]
cover = live_info["cover"]
cover = live_info["user_cover"]
sub = await BilibiliSub.get_sub(id_)
if sub.live_status != live_status:
await BilibiliSub.update_sub_info(id_, live_status=live_status)
@@ -238,15 +247,16 @@ async def _get_up_status(id_: int) -> Optional[str]:
:return:
"""
_user = await BilibiliSub.get_sub(id_)
u = user.User(_user.uid)
user_info = await u.get_user_info()
"""bilibili_api.user库中User类的get_user_info改为bilireq.user库的get_user_info方法"""
user_info = await get_user_info(id_)
uname = user_info["name"]
video_info = await u.get_videos()
"""bilibili_api.user库中User类的get_videos改为bilireq.user库的get_videos方法"""
video_info = await get_videos(id_)
latest_video_created = 0
video = None
if _user.uname != uname:
await BilibiliSub.update_sub_info(id_, uname=uname)
dynamic_img, dynamic_upload_time = await get_user_dynamic(u, _user)
dynamic_img, dynamic_upload_time = await get_user_dynamic(id_, _user)
if video_info["list"].get("vlist"):
video = video_info["list"]["vlist"][0]
latest_video_created = video["created"]
@@ -280,7 +290,8 @@ async def _get_season_status(id_) -> Optional[str]:
获取 番剧 更新状态
:param id_: 番剧 id
"""
season_info = await bangumi.get_meta(id_)
"""bilibili_api.bangumi库中get_meta改为bilireq.bangumi库的get_meta方法"""
season_info = await get_meta(id_)
title = season_info["media"]["title"]
_idx = (await BilibiliSub.get_sub(id_)).season_current_episode
new_ep = season_info["media"]["new_ep"]["index"]
@@ -297,15 +308,16 @@ async def _get_season_status(id_) -> Optional[str]:
async def get_user_dynamic(
u: User, local_user: BilibiliSub
uid: int, local_user: BilibiliSub
) -> "Optional[MessageSegment], int":
"""
获取用户动态
:param u: 用户类
:param uid: 用户uid
:param local_user: 数据库存储的用户数据
:return: 最新动态截图与时间
"""
dynamic_info = await u.get_dynamics(0)
"""bilibili_api.user库中User类的get_dynamics改为bilireq.dynamic库的get_user_dynamics方法"""
dynamic_info = await dynamic.get_user_dynamics(uid)
browser = await get_browser()
if dynamic_info.get("cards") and browser:
dynamic_upload_time = dynamic_info["cards"][0]["desc"]["timestamp"]
@@ -397,4 +409,4 @@ class SubManager:
if sub:
return sub
await self.reload_sub_data()
return await self.random_sub_data()
return await self.random_sub_data()
+48 -3
View File
@@ -1,7 +1,8 @@
from utils.image_utils import BuildImage
from configs.path_config import IMAGE_PATH
from utils.http_utils import AsyncHttpx
from bilibili_api import user
# from bilibili_api import user
from bilireq.user import get_user_info
from io import BytesIO
@@ -28,8 +29,8 @@ async def create_live_des_image(uid: int, title: str, cover: str, tags: str, des
:param des: 直播简介
:return:
"""
u = user.User(uid)
user_info = await u.get_user_info()
user_info = await get_user_info(uid)
name = user_info["name"]
sex = user_info["sex"]
face = user_info["face"]
@@ -68,3 +69,47 @@ def _create_live_des_image(
border_img = BuildImage(1772, 2657, background=border)
bk = BuildImage(1772, 2657, font_size=30)
bk.paste(cover, (0, 100), center_type="by_width")
async def get_meta(media_id: int, auth=None, reqtype="both", **kwargs):
"""
根据番剧 ID 获取番剧元数据信息,
作为bilibili_api和bilireq的替代品。
如果bilireq.bangumi更新了,可以转为调用bilireq.bangumi的get_meta方法,两者完全一致。
"""
from bilireq.utils import get
BASE_URL = "https://api.bilibili.com"
url = f"{BASE_URL}/pgc/review/user"
params = {"media_id": media_id}
raw_json = await get(url, raw=True, params=params, auth=auth, reqtype=reqtype, **kwargs)
return raw_json['result']
async def get_videos(
uid: int, tid: int = 0, pn: int = 1, keyword: str = "", order: str = "pubdate",
*, auth=None, reqtype="both", **kwargs
):
"""
获取用户投该视频信息
作为bilibili_api和bilireq的替代品。
如果bilireq.user更新了,可以转为调用bilireq.user的get_videos方法,两者完全一致。
:param uid: 用户 UID
:param tid: 分区 ID
:param pn: 页码
:param keyword: 搜索关键词
:param order: 排序方式,可以为 “pubdate(上传日期从新到旧), stow(收藏从多到少), click(播放量从多到少)”
"""
from bilireq.utils import get
BASE_URL = "https://api.bilibili.com"
url = f"{BASE_URL}/x/space/arc/search"
params = {
"mid": uid,
"ps": 30,
"tid": tid,
"pn": pn,
"keyword": keyword,
"order": order
}
return await get(url, params=params, auth=auth, reqtype=reqtype, **kwargs)
+249
View File
@@ -0,0 +1,249 @@
from nonebot.adapters.onebot.v11 import (
Event,
MessageEvent,
GroupMessageEvent,
Message,
Bot,
)
from nonebot.matcher import Matcher
from nonebot.message import run_preprocessor
from utils.image_utils import BuildImage
from utils.utils import get_message_text, is_number
from nonebot.params import RegexGroup, CommandArg
from .utils import black_word_manager
from nonebot import on_command, on_message, on_regex
from configs.config import Config, NICKNAME
from nonebot.permission import SUPERUSER
from .data_source import show_black_text_image, set_user_punish
from services.log import logger
from models.ban_user import BanUser
from datetime import datetime
from utils.message_builder import image
from .model import BlackWord
from typing import Tuple, Any
__zx_plugin_name__ = "敏感词检测"
__plugin_usage__ = """
usage:
注意你的发言!
指令:
惩罚机制
""".strip()
__plugin_superuser_usage__ = """
usage:
查看和设置惩罚
Regex:^记录名单(u:\d*)?(g:\d*)?(d[=><]\d*-\d{1,2}-\d{1,2})?$
设置惩罚id需要通过 '记录名单u:xxxxxxxx' 获取
指令:
记录名单
设置惩罚 [user_id] [id] [punish_level]
示例:记录名单
示例:记录名单u:12345678
示例:设置惩罚 12345678 1 4
""".strip()
__plugin_des__ = "请注意你的发言!!"
__plugin_type__ = ("其他",)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_cmd__ = ["惩罚机制", "记录名单 [_superuser]", "设置惩罚 [_superuser]"]
__plugin_settings__ = {
"cmd": ["敏感词检测"],
}
Config.add_plugin_config(
"black_word", "CYCLE_DAYS", 30, name="敏感词检测与惩罚", help_="黑名单词汇记录周期", default_value=30
)
Config.add_plugin_config(
"black_word",
"TOLERATE_COUNT",
[5, 1, 1, 1, 1],
help_="各个级别惩罚的容忍次数,依次为:1, 2, 3, 4, 5",
default_value=[5, 1, 1, 1, 1],
)
Config.add_plugin_config(
"black_word", "AUTO_PUNISH", True, help_="是否启动自动惩罚机制", default_value=True
)
# Config.add_plugin_config(
# "black_word", "IGNORE_GROUP", [], help_="退出群聊惩罚中忽略的群聊,即不会退出的群聊", default_value=[]
# )
Config.add_plugin_config(
"black_word",
"BAN_4_DURATION",
360,
help_="Union[int, List[int, int]]Ban时长(分钟),四级惩罚,可以为指定数字或指定列表区间(随机),例如 [30, 360]",
default_value=360,
)
Config.add_plugin_config(
"black_word",
"BAN_3_DURATION",
7,
help_="Union[int, List[int, int]]Ban时长(天),三级惩罚,可以为指定数字或指定列表区间(随机),例如 [7, 30]",
default_value=360,
)
Config.add_plugin_config(
"black_word",
"WARNING_RESULT",
f"请注意对{NICKNAME}的发言内容",
help_="口头警告内容",
default_value=f"请注意对{NICKNAME}的发言内容",
)
Config.add_plugin_config(
"black_word",
"AUTO_ADD_PUNISH_LEVEL",
True,
help_="自动提级机制,当周期内处罚次数大于某一特定值就提升惩罚等级",
default_value=True,
)
Config.add_plugin_config(
"black_word",
"ADD_PUNISH_LEVEL_TO_COUNT",
3,
help_="在CYCLE_DAYS周期内触发指定惩罚次数后提升惩罚等级",
default_value=3,
)
Config.add_plugin_config(
"black_word",
"ALAPI_CHECK_FLAG",
False,
help_="当未检测到已收录的敏感词时,开启ALAPI文本检测并将疑似文本发送给超级用户",
default_value=False,
)
Config.add_plugin_config(
"black_word",
"CONTAIN_BLACK_STOP_PROPAGATION",
True,
help_="当文本包含任意敏感词时,停止向下级插件传递,即不触发ai",
default_value=True,
)
message_matcher = on_message(priority=1, block=False)
set_punish = on_command("设置惩罚", priority=1, permission=SUPERUSER, block=True)
show_black = on_regex(
r"^记录名单(u:\d*)?(g:\d*)?(d[=><]\d*-\d{1,2}-\d{1,2})?$",
priority=1,
permission=SUPERUSER,
block=True,
)
show_punish = on_command("惩罚机制", aliases={"敏感词检测"}, priority=1, block=True)
# 黑名单词汇检测
@run_preprocessor
async def _(
bot: Bot,
matcher: Matcher,
event: Event,
):
if (
isinstance(event, MessageEvent)
and event.is_tome()
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
):
user_id = event.user_id
group_id = event.group_id if isinstance(event, GroupMessageEvent) else None
msg = get_message_text(event.json())
if await black_word_manager.check(user_id, group_id, msg) and Config.get_config(
"black_word", "CONTAIN_BLACK_STOP_PROPAGATION"
):
matcher.stop_propagation()
@show_black.handle()
async def _(bot: Bot, reg_group: Tuple[Any, ...] = RegexGroup()):
user_id, group_id, date = reg_group
date_type = "="
if date:
date_type = date[1]
date = date[2:]
try:
date = datetime.strptime(date, "%Y-%m-%d")
except ValueError:
await show_black.finish("日期格式错误,需要:年-月-日")
pic = await show_black_text_image(
bot,
int(user_id.split(":")[1]) if user_id else None,
int(group_id.split(":")[1]) if group_id else None,
date,
date_type,
)
await show_black.send(image(b64=pic.pic2bs4()))
@show_punish.handle()
async def _():
text = f"""
** 惩罚机制 **
惩罚前包含容忍机制,在指定周期内会容忍偶尔少次数的敏感词只会进行警告提醒
多次触发同级惩罚会使惩罚等级提高,即惩罚自动提级机制
目前公开的惩罚等级:
1级:永久ban
2级:删除好友
3级:ban指定/随机天数
4级:ban指定/随机时长
5级:警告
备注:
该功能为测试阶段,如果你有被误封情况,请联系管理员,会从数据库中提取出你的数据进行审核后判断
目前该功能暂不完善,部分情况会由管理员鉴定,请注意对真寻的发言
关于敏感词:
记住不要骂{NICKNAME}就对了!
""".strip()
max_width = 0
for m in text.split("\n"):
max_width = len(m) * 20 if len(m) * 20 > max_width else max_width
max_height = len(text.split("\n")) * 24
A = BuildImage(
max_width, max_height, font="CJGaoDeGuo.otf", font_size=24, color="#E3DBD1"
)
A.text((10, 10), text)
await show_punish.send(image(b64=A.pic2bs4()))
@set_punish.handle()
async def _(event: MessageEvent, arg: Message = CommandArg()):
msg = arg.extract_plain_text().strip().split()
if (
len(msg) < 3
or not is_number(msg[0])
or not is_number(msg[1])
or not is_number(msg[2])
):
await set_punish.finish("参数错误,请查看帮助...", at_sender=True)
uid = int(msg[0])
id_ = int(msg[1])
punish_level = int(msg[2])
print(uid, id_, punish_level)
rst = await set_user_punish(uid, id_, punish_level)
await set_punish.send(rst)
logger.info(
f"USER {event.user_id} 设置惩罚 uid:{uid} id_:{id_} punish_level:{punish_level} --> {rst}"
)
+118
View File
@@ -0,0 +1,118 @@
from nonebot.adapters.onebot.v11 import Bot
from utils.image_utils import BuildImage, text2image
from services.log import logger
from typing import Optional
from datetime import datetime
from .model import BlackWord
from .utils import _get_punish, Config
async def show_black_text_image(
bot: Bot,
user: Optional[int],
group_id: Optional[int],
date: Optional[datetime],
data_type: str = "=",
) -> BuildImage:
"""
展示记录名单
:param bot: bot
:param user: 用户qq
:param group_id: 群聊
:param date: 日期
:param data_type: 日期搜索类型
:return:
"""
data = await BlackWord.get_black_data(user, group_id, date, data_type)
A = BuildImage(0, 0, color="#f9f6f2", font_size=20)
image_list = []
friend_str = await bot.get_friend_list()
id_str = ""
uname_str = ""
uid_str = ""
gid_str = ""
plant_text_str = ""
black_word_str = ""
punish_str = ""
punish_level_str = ""
create_time_str = ""
for i, x in enumerate(data):
try:
if x.group_id:
user_name = (
await bot.get_group_member_info(
group_id=x.group_id, user_id=x.user_qq
)
)["card"]
else:
user_name = [
u["nickname"] for u in friend_str if u["user_id"] == x.user_qq
][0]
except Exception as e:
logger.warning(
f"show_black_text_image 获取 USER {x.user_qq} user_name 失败 {type(e)}:{e}"
)
user_name = x.user_qq
id_str += f"{i}\n"
uname_str += f"{user_name}\n"
uid_str += f"{x.user_qq}\n"
gid_str += f"{x.group_id}\n"
plant_text = " ".join(x.plant_text.split("\n"))
if A.getsize(plant_text)[0] > 200:
plant_text = plant_text[:20] + "..."
plant_text_str += f"{plant_text}\n"
black_word_str += f"{x.black_word}\n"
punish_str += f"{x.punish}\n"
punish_level_str += f"{x.punish_level}\n"
create_time_str += f"{x.create_time.replace(microsecond=0)}\n"
_tmp_img = BuildImage(0, 0, font_size=35, font="CJGaoDeGuo.otf")
for s, type_ in [
(id_str, "Id"),
(uname_str, "昵称"),
(uid_str, "UID"),
(gid_str, "GID"),
(plant_text_str, "文本"),
(black_word_str, "检测"),
(punish_str, "惩罚"),
(punish_level_str, "等级"),
(create_time_str, "记录日期"),
]:
img = await text2image(s, color="#f9f6f2", _add_height=3.32)
w = _tmp_img.getsize(type_)[0] if _tmp_img.getsize(type_)[0] > img.w else img.w
A = BuildImage(w + 11, img.h + 50, color="#f9f6f2", font_size=35, font="CJGaoDeGuo.otf")
await A.atext((10, 10), type_)
await A.apaste(img, (0, 50))
image_list.append(A)
horizontal_line = []
w, h = 0, 0
for img in image_list:
w += img.w + 20
h = img.h if img.h > h else h
horizontal_line.append(img.w)
A = BuildImage(w, h, color="#f9f6f2")
current_w = 0
for img in image_list:
await A.apaste(img, (current_w, 0))
current_w += img.w + 20
return A
async def set_user_punish(user_id: int, id_: int, punish_level: int) -> str:
"""
设置惩罚
:param user_id: 用户id
:param id_: 记录下标
:param punish_level: 惩罚等级
"""
result = await _get_punish(punish_level, user_id)
punish = {
1: "永久ban",
2: "删除好友",
3: f"ban {result} 天",
4: f"ban {result} 分钟",
5: "口头警告"
}
if await BlackWord.set_user_punish(user_id, punish[punish_level], id_=id_):
return f"已对 USER {user_id} 进行 {punish[punish_level]} 处罚。"
else:
return "操作失败,可能未找到用户,id或敏感词"
+149
View File
@@ -0,0 +1,149 @@
from services.db_context import db
from typing import Optional, List
from datetime import datetime, timedelta
class BlackWord(db.Model):
__tablename__ = "black_word"
id = db.Column(db.Integer(), primary_key=True, autoincrement=True)
user_qq = db.Column(db.BigInteger(), nullable=False, primary_key=True)
group_id = db.Column(db.BigInteger())
plant_text = db.Column(db.String())
black_word = db.Column(db.String())
punish = db.Column(db.String(), default="")
punish_level = db.Column(db.Integer())
create_time = db.Column(db.DateTime(timezone=True), nullable=False)
@classmethod
async def add_user_black_word(
cls,
user_qq: int,
group_id: Optional[int],
black_word: str,
plant_text: str,
punish_level: int,
):
"""
说明:
添加用户发送的敏感词
参数:
:param user_qq: 用户id
:param group_id: 群号
:param black_word: 黑名单词汇
:param plant_text: 消息文本
:param punish_level: 惩罚等级
"""
await cls.create(
user_qq=user_qq,
group_id=group_id,
plant_text=plant_text,
black_word=black_word,
punish_level=punish_level,
create_time=datetime.now(),
)
@classmethod
async def set_user_punish(
cls,
user_qq: int,
punish: str,
black_word: Optional[str] = None,
id_: Optional[int] = None,
) -> bool:
"""
说明:
设置处罚
参数:
:param user_qq: 用户id
:param punish: 处罚
:param black_word: 黑名单词汇
:param id_: 记录下标
"""
user = None
if (not black_word and not id_) or not punish:
return False
query = cls.query.where(cls.user_qq == user_qq).with_for_update()
if black_word:
user = await query.where(cls.black_word == black_word).order_by(cls.id.desc()).gino.first()
elif id_:
user_list = await query.gino.all()
print(len(user_list))
if len(user_list) == 0 or (id_ < 0 or id_ > len(user_list)):
return False
user = user_list[id_]
if not user:
return False
await user.update(punish=cls.punish + punish + " ").apply()
return True
@classmethod
async def get_user_count(
cls, user_qq: int, days: int = 7, punish_level: Optional[int] = None
) -> int:
"""
说明:
获取用户规定周期内的犯事次数
参数:
:param user_qq: 用户qq
:param days: 周期天数
:param punish_level: 惩罚等级
"""
setattr(BlackWord, "count", db.func.count(cls.id).label("count"))
query = cls.select("count").where(
(cls.user_qq == user_qq)
& (cls.punish_level != -1)
& (cls.create_time > datetime.now() - timedelta(days=days))
)
if punish_level is not None:
query = query.where(cls.punish_level == punish_level)
return (await query.gino.first())[0]
@classmethod
async def get_user_punish_level(cls, user_qq: int, days: int = 7) -> Optional[int]:
"""
说明:
获取用户最近一次的惩罚记录等级
参数:
:param user_qq: 用户qq
:param days: 周期天数
"""
if (
query := await cls.query.where(cls.user_qq == user_qq)
.where(cls.create_time > datetime.now() - timedelta(days=days))
.order_by(cls.id.desc())
.gino.first()
):
return query.punish_level
return None
@classmethod
async def get_black_data(
cls,
user_qq: Optional[int],
group_id: Optional[int],
date: Optional[datetime],
date_type: str = "=",
) -> List["BlackWord"]:
"""
说明:
通过指定条件查询数据
参数:
:param user_qq: 用户qq
:param group_id: 群号
:param date: 日期
:param date_type: 日期查询类型
"""
query = cls.query
if user_qq:
query = query.where(cls.user_qq == user_qq)
if group_id:
query = query.where(cls.group_id == group_id)
if date:
if date_type == "=":
query = query.where(cls.create_time == date)
elif date_type == ">":
query = query.where(cls.create_time > date)
elif date_type == "<":
query = query.where(cls.create_time < date)
return await query.gino.all()
+324
View File
@@ -0,0 +1,324 @@
from utils.utils import cn2py, get_bot
from configs.path_config import DATA_PATH
from typing import Optional, Union, Tuple
from .model import BlackWord
from configs.config import Config
from pathlib import Path
from services.log import logger
from models.ban_user import BanUser
from nonebot.adapters.onebot.v11.exception import ActionFailed
from models.group_member_info import GroupInfoUser
from utils.http_utils import AsyncHttpx
import random
try:
import ujson as json
except ModuleNotFoundError:
import json
class BlackWordManager:
"""
敏感词管理( 拒绝恶意
"""
def __init__(self, word_file: Path, py_file: Path):
self._word_list = {
"1": [],
"2": [],
"3": [],
"4": ["sb", "nmsl", "mdzz", "2b", "jb", "操", "废物", "憨憨", "cnm", "rnm"],
"5": [],
}
self._py_list = {
"1": [],
"2": [],
"3": [],
"4": [
"shabi",
"wocaonima",
"sima",
"sabi",
"zhizhang",
"naocan",
"caonima",
"rinima",
"simadongxi",
"simawanyi",
"hanbi",
"hanpi",
"laji",
"fw"
],
"5": [],
}
word_file.parent.mkdir(parents=True, exist_ok=True)
if word_file.exists():
# 清空默认配置
with open(word_file, "r", encoding="utf8") as f:
self._word_list = json.load(f)
else:
with open(word_file, "w", encoding="utf8") as f:
json.dump(
self._word_list,
f,
ensure_ascii=False,
indent=4,
)
if py_file.exists():
# 清空默认配置
with open(py_file, "r", encoding="utf8") as f:
self._py_list = json.load(f)
else:
with open(py_file, "w", encoding="utf8") as f:
json.dump(
self._py_list,
f,
ensure_ascii=False,
indent=4,
)
async def check(
self, user_id: int, group_id: Optional[int], message: str
) -> Optional[Union[str, bool]]:
"""
检查是否包含黑名单词汇
:param user_id: 用户id
:param group_id: 群号
:param message: 消息
"""
print(user_id, group_id, message)
if data := self._check(message):
print(data)
if data[0]:
await _add_user_black_word(
user_id, group_id, data[0], message, int(data[1])
)
return True
if Config.get_config(
"black_word", "ALAPI_CHECK_FLAG"
) and not await check_text(message):
await send_msg(
0, None, f"USER {user_id} GROUP {group_id} ALAPI 疑似检测:{message}"
)
return False
def _check(self, message: str) -> Tuple[Optional[str], int]:
"""
检测文本是否违规
:param message: 检测消息
"""
# 移除空格
message = message.replace(" ", "")
py_msg = cn2py(message).lower()
# 完全匹配
for x in [self._word_list, self._py_list]:
for level in x:
if message in x[level] or py_msg in x[level]:
return message if message in x[level] else py_msg, level
# 模糊匹配
for x in [self._word_list, self._py_list]:
for level in x:
for m in x[level]:
if m in message or m in py_msg:
return m, -1
return None, 0
async def _add_user_black_word(
user_id: int,
group_id: Optional[int],
black_word: str,
message: str,
punish_level: int,
):
"""
添加敏感词数据
:param user_id: 用户id
:param group_id: 群号
:param black_word: 触发的黑名单词汇
:param message: 原始文本
:param punish_level: 惩罚等级
"""
cycle_days = Config.get_config("black_word", "CYCLE_DAYS") or 7
user_count = await BlackWord.get_user_count(user_id, cycle_days, punish_level)
# 周期内超过次数直接提升惩罚
if Config.get_config(
"black_word", "AUTO_ADD_PUNISH_LEVEL"
) and user_count > Config.get_config("black_word", "ADD_PUNISH_LEVEL_TO_COUNT"):
punish_level -= 1
await BlackWord.add_user_black_word(
user_id, group_id, black_word, message, punish_level
)
logger.info(
f"已将 USER {user_id} GROUP {group_id} 添加至黑名单词汇记录 Black_word:{black_word} Plant_text:{message}"
)
# 自动惩罚
if Config.get_config("black_word", "AUTO_PUNISH") and punish_level != -1:
await _punish_handle(user_id, group_id, punish_level, black_word)
async def _punish_handle(
user_id: int, group_id: Optional[int], punish_level: int, black_word: str
):
"""
惩罚措施,级别越低惩罚越严
:param user_id: 用户id
:param group_id: 群号
:param black_word: 触发的黑名单词汇
"""
logger.info(f"BlackWord USER {user_id} 触发 {punish_level} 级惩罚...")
# 周期天数
cycle_days = Config.get_config("black_word", "CYCLE_DAYS") or 7
# 用户周期内触发punish_level级惩罚的次数
user_count = await BlackWord.get_user_count(user_id, cycle_days, punish_level)
# 获取最近一次的惩罚等级,将在此基础上增加
punish_level = await BlackWord.get_user_punish_level(user_id, cycle_days) or punish_level
# 容忍次数:List[int]
tolerate_count = Config.get_config("black_word", "TOLERATE_COUNT")
if not tolerate_count or len(tolerate_count) < 5:
tolerate_count = [5, 2, 2, 2, 2]
if punish_level == 1 and user_count > tolerate_count[punish_level - 1]:
# 永久ban
await _get_punish(1, user_id, group_id)
await BlackWord.set_user_punish(user_id, "永久ban 删除好友", black_word)
elif punish_level == 2 and user_count > tolerate_count[punish_level - 1]:
# 删除好友
await _get_punish(2, user_id, group_id)
await BlackWord.set_user_punish(user_id, "删除好友", black_word)
elif punish_level == 3 and user_count > tolerate_count[punish_level - 1]:
# 永久ban
ban_day = await _get_punish(3, user_id, group_id)
await BlackWord.set_user_punish(user_id, f"ban {ban_day} 天", black_word)
elif punish_level == 4 and user_count > tolerate_count[punish_level - 1]:
# ban指定时长
ban_time = await _get_punish(4, user_id, group_id)
await BlackWord.set_user_punish(user_id, f"ban {ban_time} 分钟", black_word)
elif punish_level == 5 and user_count > tolerate_count[punish_level - 1]:
# 口头警告
warning_result = await _get_punish(5, user_id, group_id)
await BlackWord.set_user_punish(user_id, f"口头警告:{warning_result}", black_word)
else:
await BlackWord.set_user_punish(user_id, f"提示!", black_word)
await send_msg(
user_id,
group_id,
f"BlackWordChecker:该条发言已被记录,目前你在{cycle_days}天内的发表{punish_level}级"
f"言论记录次数为:{user_count}次,请注意你的发言\n"
f"* 如果你不清楚惩罚机制,请发送“惩罚机制” *",
)
async def _get_punish(
id_: int, user_id: int, group_id: Optional[int] = None
) -> Optional[Union[int, str]]:
"""
通过id_获取惩罚
:param id_: id
:param user_id: 用户id
:param group_id: 群号
"""
bot = get_bot()
# 忽略的群聊
# _ignore_group = Config.get_config("black_word", "IGNORE_GROUP")
# 处罚 id 4 ban 时间:int,List[int]
ban_3_duration = Config.get_config("black_word", "BAN_3_DURATION")
# 处罚 id 4 ban 时间:int,List[int]
ban_4_duration = Config.get_config("black_word", "BAN_4_DURATION")
# 口头警告内容
warning_result = Config.get_config("black_word", "WARNING_RESULT")
try:
uname = (await GroupInfoUser.get_member_info(user_id, group_id)).user_name
except AttributeError:
uname = user_id
# 永久ban
if id_ == 1:
if str(user_id) not in bot.config.superusers:
await BanUser.ban(user_id, 10, 99999999)
await send_msg(user_id, group_id, f"BlackWordChecker 永久ban USER {uname}({user_id})")
logger.info(f"BlackWord 永久封禁 USER {user_id}...")
# 删除好友(有的话
elif id_ == 2:
if str(user_id) not in bot.config.superusers:
try:
await bot.delete_friend(user_id=user_id)
await send_msg(
user_id, group_id, f"BlackWordChecker 删除好友 USER {uname}({user_id})"
)
logger.info(f"BlackWord 删除好友 {user_id}...")
except ActionFailed:
pass
# 封禁用户指定时间,默认7天
elif id_ == 3:
if isinstance(ban_3_duration, list):
ban_3_duration = random.randint(ban_3_duration[0], ban_3_duration[1])
await BanUser.ban(user_id, 9, ban_4_duration * 60 * 60 * 24)
await send_msg(
user_id,
group_id,
f"BlackWordChecker 对用户 USER {uname}({user_id}) 进行封禁 {ban_3_duration} 天处罚。",
)
logger.info(f"BlackWord 封禁 USER {uname}({user_id}) {ban_3_duration} 天...")
return ban_3_duration
# 封禁用户指定时间,默认360分钟
elif id_ == 4:
if isinstance(ban_4_duration, list):
ban_4_duration = random.randint(ban_4_duration[0], ban_4_duration[1])
await BanUser.ban(user_id, 9, ban_4_duration * 60)
await send_msg(
user_id,
group_id,
f"BlackWordChecker 对用户 USER {uname}({user_id}) 进行封禁 {ban_4_duration} 分钟处罚。",
)
logger.info(f"BlackWord 封禁 USER {uname}({user_id}) {ban_4_duration} 分钟...")
return ban_4_duration
# 口头警告
elif id_ == 5:
if group_id:
await bot.send_group_msg(group_id=group_id, message=warning_result)
else:
await bot.send_private_msg(user_id=user_id, message=warning_result)
logger.info(f"BlackWord 口头警告 USER {user_id}")
return warning_result
return None
async def send_msg(user_id: int, group_id: Optional[int], message: str):
"""
发送消息
:param user_id: user_id
:param group_id: group_id
:param message: message
"""
bot = get_bot()
if not user_id:
user_id = int(list(bot.config.superusers)[0])
if group_id:
await bot.send_group_msg(group_id=group_id, message=message)
else:
await bot.send_private_msg(user_id=user_id, message=message)
async def check_text(text: str) -> bool:
"""
ALAPI文本检测,检测输入违规
:param text: 回复
"""
if not Config.get_config("alapi", "ALAPI_TOKEN"):
return True
params = {"token": Config.get_config("alapi", "ALAPI_TOKEN"), "text": text}
try:
data = (
await AsyncHttpx.get(
"https://v2.alapi.cn/api/censor/text", timeout=4, params=params
)
).json()
if data["code"] == 200:
return data["data"]["conclusion_type"] == 2
except Exception as e:
logger.error(f"检测违规文本错误...{type(e)}:{e}")
return True
black_word_manager = BlackWordManager(DATA_PATH / "black_word" / "black_word.json", DATA_PATH / "black_word" / "black_py.json")
@@ -73,6 +73,7 @@ async def check_update(bot: Bot) -> 'int, str':
message=f"检测真寻已更新,当前版本:{_version},最新版本:{latest_version}\n" f"开始更新.....",
)
logger.info(f"开始下载真寻最新版文件....")
tar_gz_url = (await AsyncHttpx.get(tar_gz_url)).headers.get('Location')
if await AsyncHttpx.download_file(tar_gz_url, zhenxun_latest_tar_gz):
logger.info("下载真寻最新版文件完成....")
error = await asyncio.get_event_loop().run_in_executor(
@@ -256,9 +256,7 @@ class Genshin(db.Model):
if x:
await cls._add_query_uid(uid, uid)
return x.cookie
for u in [
x for x in await cls.query.order_by(db.func.random()).gino.all() if x.cookie
]:
for u in await cls.query.where(cls.cookie != "").order_by(db.func.random()).gino.all():
if not u.today_query_uid or len(u.today_query_uid[:-1].split()) < 30:
await cls._add_query_uid(uid, u.uid)
return u.cookie
@@ -23,17 +23,17 @@ async def _():
g_list = await Genshin.get_all_auto_sign_user()
for u in g_list:
if u.auto_sign_time:
date = await Genshin.random_sign_time(u.uid)
scheduler.add_job(
_sign,
"date",
run_date=date.replace(microsecond=0),
id=f"genshin_auto_sign_{u.uid}_{u.user_qq}_0",
args=[u.user_qq, u.uid, 0],
)
logger.info(
f"genshin_sign add_job:USER:{u.user_qq} UID:{u.uid} " f"{date} 原神自动签到"
)
if date := await Genshin.random_sign_time(u.uid):
scheduler.add_job(
_sign,
"date",
run_date=date.replace(microsecond=0),
id=f"genshin_auto_sign_{u.uid}_{u.user_qq}_0",
args=[u.user_qq, u.uid, 0],
)
logger.info(
f"genshin_sign add_job:USER:{u.user_qq} UID:{u.uid} " f"{date} 原神自动签到"
)
def add_job(user_id: int, uid: int, date: datetime):
@@ -299,8 +299,7 @@ def get_country_data_image(world_data_dict: Dict) -> BuildImage:
# 层岩巨渊 和 地下矿区 算一个
region = BuildImage(790, 267 * (len(world_data_dict) - 1), color="#F9F6F2")
height = 0
for country in ["蒙德", "龙脊雪山", "璃月", "璃月层岩巨渊", "稻妻", "渊下宫"]:
for country in ["蒙德", "龙脊雪山", "璃月", "层岩巨渊", "稻妻", "渊下宫"]:
x = BuildImage(790, 250, color="#3A4467")
logo = BuildImage(180, 180, background=image_path / "logo" / f"{country}.png")
tmp_bk = BuildImage(770, 230, color="#606779")
@@ -322,23 +321,23 @@ def get_country_data_image(world_data_dict: Dict) -> BuildImage:
f"Lv.{world_data_dict[country]['level']}",
fill=(255, 255, 255),
)
elif country in ["璃月层岩巨渊"]:
elif country in ["层岩巨渊"]:
content_bk.text((300, 20), "层岩巨渊探索", fill=(239, 211, 114))
content_bk.text(
(570, 20),
f"{world_data_dict['璃月层岩巨渊']['exploration_percentage'] / 10}%",
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}%",
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']}",
f"LV.{world_data_dict['层岩巨渊·地下矿区']['offerings'][0]['level']}",
fill=(255, 255, 255),
)
elif country in ["龙脊雪山"]:
@@ -20,6 +20,9 @@ driver: Driver = nonebot.get_driver()
get_memo = require("query_memo").get_memo
global_map = {}
class UserManager:
def __init__(self, max_error_count: int = 3):
self._data = []
@@ -146,8 +149,8 @@ async def _remind(user_id: int, uid: str):
if current_resin < max_resin:
user_manager.remove(uid)
user_manager.remove_overflow(uid)
if max_resin - 40 <= current_resin <= max_resin - 20:
next_time = now + timedelta(minutes=(max_resin - 20 - current_resin) * 8, seconds=10)
if max_resin - 40 < current_resin <= max_resin - 20:
next_time = now + timedelta(minutes=(max_resin - 20 - current_resin + 1) * 8, seconds=10)
elif current_resin < max_resin:
next_time = now + timedelta(minutes=(max_resin - current_resin) * 8, seconds=10)
elif current_resin == max_resin:
@@ -189,6 +192,7 @@ async def _remind(user_id: int, uid: str):
user_manager.remove_error_count(uid)
await Genshin.set_user_resin_recovery_time(int(uid), next_time)
scheduler.add_job(
_remind,
_remind,
"date",
run_date=next_time,
-34
View File
@@ -1,34 +0,0 @@
from nonebot import on_message
from nonebot.adapters.onebot.v11.permission import GROUP
from nonebot.adapters.onebot.v11 import GroupMessageEvent
from .data_source import cancel_all_notice, save_data, get_data, set_data_value
from services.log import logger
import time
__zx_plugin_name__ = "群聊最后聊天时间记录 [Hidden]"
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
last_chat = on_message(priority=1, block=False, permission=GROUP)
@last_chat.handle()
async def _(event: GroupMessageEvent):
time_data = await get_data()
set_data_value(event.group_id, time.time())
if event.group_id in time_data["_group"]:
time_data["_group"].remove(event.group_id)
set_data_value("_group", time_data["_group"])
for key in time_data.keys():
if key not in ["check_time", "_group"]:
if key not in time_data["_group"]:
if time.time() - time_data[key] > 60 * 60 * 36:
await cancel_all_notice(key)
time_data["_group"].append(key)
set_data_value("_group", time_data["_group"])
logger.info(f"GROUP {event.group_id} 因群内发言时间大于36小时被取消全部通知")
if time.time() - time_data["check_time"] > 60 * 60 * 1:
set_data_value("check_time", time.time())
save_data()
-67
View File
@@ -1,67 +0,0 @@
from configs.path_config import DATA_PATH
from utils.utils import get_bot
from datetime import datetime
import time
from services.log import logger
from utils.manager import group_manager
try:
import ujson as json
except ModuleNotFoundError:
import json
time_data = {}
async def init():
global time_data
bot = get_bot()
gl = await bot.get_group_list()
gl = [g["group_id"] for g in gl]
data = read_data("group_last_chat_time.json")
for g in gl:
if not data.get(g):
time_data[g] = time.time()
if not time_data.get("check_time"):
time_data["check_time"] = time.time()
if not time_data.get("_group"):
time_data["_group"] = []
save_data()
return time_data
def read_data(file_name: str):
try:
with open(DATA_PATH / file_name, "r", encoding="utf8") as f:
return json.load(f)
except (ValueError, FileNotFoundError):
return {}
def save_data():
with open(DATA_PATH / "group_last_chat_time.json", "w") as f:
json.dump(time_data, f, indent=4)
logger.info(
f'自动存储 group_last_chat_time.json 时间:{str(datetime.now()).split(".")[0]}'
)
# 取消全部通知
async def cancel_all_notice(group_id):
group_id = int(group_id)
for command in group_manager.get_task_data():
if await group_manager.check_group_task_status(group_id, command):
await group_manager.close_group_task(group_id, command)
logger.info(f"关闭了 {group_id} 群的全部通知")
async def get_data():
global time_data
if not time_data:
time_data = await init()
return time_data
def set_data_value(key, value):
global time_data
time_data[key] = value
+2 -2
View File
@@ -64,8 +64,8 @@ async def _poke_event(event: PokeNotifyEvent):
rand = random.random()
if rand <= 0.3:
path = random.choice(["luoli", "meitu"])
index = random.randint(0, len(os.listdir(IMAGE_PATH / path)))
result = f"id:{index}" + image(f"{index}.jpg", path)
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}")
elif 0.3 < rand < 0.6:
-1
View File
@@ -144,7 +144,6 @@ class Setu(db.Model):
return _tmp_local_id
return -1
@classmethod
async def update_setu_data(
cls,
@@ -139,7 +139,7 @@ async def update_setu_img(flag: bool = False):
f"--> /{path}/{image.local_id}.jpg"
)
os.rename(
TEMP_PATH / f"/{image.local_id}.jpg",
TEMP_PATH / f"{image.local_id}.jpg",
path / f"{image.local_id}.jpg",
)
except FileNotFoundError:
+7 -5
View File
@@ -1,3 +1,5 @@
from typing import Tuple, Any
from .group_user_checkin import (
group_user_check_in,
group_user_check,
@@ -8,9 +10,9 @@ from .group_user_checkin import (
from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message
from nonebot.adapters.onebot.v11.permission import GROUP
from utils.message_builder import image
from nonebot import on_command
from nonebot import on_command, on_regex
from utils.utils import scheduler
from nonebot.params import CommandArg
from nonebot.params import CommandArg, RegexGroup
from pathlib import Path
from configs.path_config import DATA_PATH
from services.log import logger
@@ -68,7 +70,7 @@ except (FileNotFoundError, ValueError, TypeError):
data = {"0": []}
sign = on_command("签到", priority=5, permission=GROUP, block=True)
sign = on_regex("^签到(all)?$", priority=5, permission=GROUP, block=True)
my_sign = on_command(
cmd="我的签到", aliases={"好感度"}, priority=5, permission=GROUP, block=True
)
@@ -85,13 +87,13 @@ total_sign_rank = on_command(
@sign.handle()
async def _(event: GroupMessageEvent, arg: Message = CommandArg()):
async def _(event: GroupMessageEvent, reg_group: Tuple[Any, ...] = RegexGroup()):
nickname = event.sender.card or event.sender.nickname
await sign.send(
await group_user_check_in(nickname, event.user_id, event.group_id),
at_sender=True,
)
if arg.extract_plain_text().strip() == "all":
if reg_group[0]:
await check_in_all(nickname, event.user_id)
+1 -1
View File
@@ -53,7 +53,7 @@ __plugin_cmd__ = [
"我的周功能调用统计 ?[功能]",
"我的月功能调用统计 ?[功能]",
]
__plugin_type__ = ("功能调用统计可视化", 1)
__plugin_type__ = ("数据统计", 1)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
+4 -206
View File
@@ -1,206 +1,4 @@
from utils.manager import (
plugins_manager,
group_manager,
plugins2settings_manager,
plugins2cd_manager,
plugins2block_manager,
plugins2count_manager,
requests_manager,
)
from ..auth import token_to_user, Depends, User
from utils.utils import get_matchers, get_bot
from models.group_info import GroupInfo
from pydantic.error_wrappers import ValidationError
from services.log import logger
from ..config import *
import nonebot
app = nonebot.get_app()
plugin_name_list = None
@app.get("/webui/plugins")
def _(type_: Optional[str], user: User = Depends(token_to_user)) -> Result:
"""
获取插件列表
:param type_: 类型 normal, superuser, hidden, admin
"""
global plugin_name_list
if not plugin_name_list:
plugin_name_list = [x.plugin_name for x in get_matchers()]
plugin_list = []
plugin_data = plugins_manager.get_data()
for model in plugin_data:
if model in plugin_name_list:
data = plugin_data.get(model)
data["model"] = model
plugin_name = data.get("plugin_name")
if (
(type_ == "hidden" and "[hidden]" not in plugin_name.lower())
or (type_ == "admin" and "[admin]" not in plugin_name.lower())
or (type_ == "superuser" and "[superuser]" not in plugin_name.lower())
):
continue
if type_ == "normal" and (
"[hidden]" in plugin_name.lower()
or "[admin]" in plugin_name.lower()
or "[superuser]" in plugin_name.lower()
):
continue
data = {"model": model}
if x := plugin_data.get(model):
if not x.get("status") and x.get("block_type") in [
"group",
"private",
"all",
]:
x["block_type"] = (
"群聊"
if x["block_type"] == "group"
else "私聊"
if x["block_type"] == "private"
else "全部"
)
data["plugin_manager"] = PluginManager(**x)
if x := plugins2settings_manager.get(model):
if x.get("cmd") and isinstance(x.get("cmd"), list):
x["cmd"] = ",".join(x["cmd"])
# if isinstance(x["plugin_type"], list):
# x["plugin_type"] = x["plugin_type"][0]
data["plugin_settings"] = PluginSettings(**x)
if x := plugins2cd_manager.get(model):
data["cd_limit"] = CdLimit(**x)
if x := plugins2block_manager.get(model):
data["block_limit"] = BlockLimit(**x)
if x := plugins2count_manager.get(model):
data["count_limit"] = CountLimit(**x)
# if x := resources_manager.get(model):
# data = dict(data, **x)
plugin_list.append(Plugin(**data))
return Result(code=200, data=plugin_list)
@app.post("/webui/plugins")
def _(plugin: Plugin, user: User = Depends(token_to_user)) -> Result:
"""
修改插件信息
:param plugin: 插件内容
"""
if plugin.plugin_settings:
for key, value in plugin.plugin_settings:
plugins2settings_manager.set_module_data(plugin.model, key, value)
if plugin.plugin_manager:
for key, value in plugin.plugin_manager:
plugins_manager.set_module_data(plugin.model, key, value)
return Result(code=200)
@app.get("/webui/group")
async def _(user: User = Depends(token_to_user)) -> Result:
"""
获取群信息
"""
group_list_result = []
group_info = {}
if bot := get_bot():
group_list = await bot.get_group_list()
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"]:
try:
task_list = []
data = group_data["group_manager"][group_id]
for tn, status in data["group_task_status"].items():
task_list.append(
Task(
**{
"name": tn,
"nameZh": group_manager.get_task_data().get(tn) or tn,
"status": status,
}
)
)
data["task"] = task_list
if x := group_info.get(int(group_id)):
data["group"] = x
else:
continue
try:
group_list_result.append(GroupResult(**data))
except ValidationError:
pass
except Exception as e:
logger.error(f"WEB_UI /webui/group 发生错误 {type(e)}:{e}")
return Result(code=200, data=group_list_result)
@app.post("/webui/group")
async def _(group: GroupResult, user: User = Depends(token_to_user)) -> Result:
"""
修改群信息
"""
group_id = group.group.group_id
group_manager.set_group_level(group_id, group.level)
if group.status:
group_manager.turn_on_group_bot_status(group_id)
else:
group_manager.shutdown_group_bot_status(group_id)
return Result(code=200)
@app.get("/webui/request")
def _(type_: Optional[str], user: User = Depends(token_to_user)) -> Result:
req_data = requests_manager.get_data()
req_list = []
if type_ in ["group", "private"]:
req_data = req_data[type_]
for x in req_data:
req_data[x]["oid"] = x
req_list.append(RequestResult(**req_data[x]))
return Result(code=200, data=req_list)
@app.delete("/webui/request")
def _(type_: Optional[str], user: User = Depends(token_to_user)) -> Result:
"""
清空请求
:param type_: 类型
"""
requests_manager.clear(type_)
return Result(code=200)
@app.post("/webui/request")
async def _(parma: RequestParma, user: User = Depends(token_to_user)) -> Result:
"""
操作请求
:param parma: 参数
"""
result = "error"
if bot := get_bot():
if parma.handle == "approve":
if parma.type == "group":
rid = requests_manager.get_group_id(parma.id)
if await GroupInfo.get_group_info(rid):
await GroupInfo.set_group_flag(rid, 1)
else:
group_info = await bot.get_group_info(group_id=rid)
await GroupInfo.add_group_info(
rid,
group_info["group_name"],
group_info["max_member_count"],
group_info["member_count"],
1,
)
if await requests_manager.approve(bot, parma.id, parma.type):
result = "ok"
elif parma.handle == "refuse":
if await requests_manager.refused(bot, parma.id, parma.type):
result = "ok"
elif parma.handle == "delete":
requests_manager.delete_request(parma.id, parma.type)
result = "ok"
return Result(code=200, data=result)
from .group import *
from .plugins import *
from .request import *
from .system import *
+61
View File
@@ -0,0 +1,61 @@
from pydantic.error_wrappers import ValidationError
from services.log import logger
from utils.manager import group_manager
from utils.utils import get_bot
from ..auth import Depends, User, token_to_user
from ..config import *
@app.get("/webui/group")
async def _(user: User = Depends(token_to_user)) -> Result:
"""
获取群信息
"""
group_list_result = []
group_info = {}
if bot := get_bot():
group_list = await bot.get_group_list()
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"]:
try:
task_list = []
data = group_data["group_manager"][group_id]
for tn, status in data["group_task_status"].items():
task_list.append(
Task(
**{
"name": tn,
"nameZh": group_manager.get_task_data().get(tn) or tn,
"status": status,
}
)
)
data["task"] = task_list
if x := group_info.get(int(group_id)):
data["group"] = x
else:
continue
try:
group_list_result.append(GroupResult(**data))
except ValidationError:
pass
except Exception as e:
logger.error(f"WEB_UI /webui/group 发生错误 {type(e)}:{e}")
return Result(code=200, data=group_list_result)
@app.post("/webui/group")
async def _(group: GroupResult, user: User = Depends(token_to_user)) -> Result:
"""
修改群信息
"""
group_id = group.group.group_id
group_manager.set_group_level(group_id, group.level)
if group.status:
group_manager.turn_on_group_bot_status(group_id)
else:
group_manager.shutdown_group_bot_status(group_id)
return Result(code=200)
+142
View File
@@ -0,0 +1,142 @@
from configs.config import Config
from services.log import logger
from utils.manager import (plugins2block_manager, plugins2cd_manager,
plugins2count_manager, plugins2settings_manager,
plugins_manager)
from utils.utils import get_matchers
from ..auth import Depends, User, token_to_user
from ..config import *
plugin_name_list = None
@app.get("/webui/plugins")
def _(type_: Optional[str], user: User = Depends(token_to_user)) -> Result:
"""
获取插件列表
:param type_: 类型 normal, superuser, hidden, admin
"""
global plugin_name_list
if not plugin_name_list:
plugin_name_list = [x.plugin_name for x in get_matchers()]
plugin_list = []
plugin_data = plugins_manager.get_data()
for model in plugin_data:
if model in plugin_name_list:
try:
data = plugin_data.get(model)
data["model"] = model
plugin_name = data.get("plugin_name")
if (
(type_ == "hidden" and "[hidden]" not in plugin_name.lower())
or (type_ == "admin" and "[admin]" not in plugin_name.lower())
or (
type_ == "superuser"
and "[superuser]" not in plugin_name.lower()
)
):
continue
if type_ == "normal" and (
"[hidden]" in plugin_name.lower()
or "[admin]" in plugin_name.lower()
or "[superuser]" in plugin_name.lower()
):
continue
data = {"model": model}
if x := plugin_data.get(model):
if not x.get("status") and x.get("block_type") in [
"group",
"private",
"all",
]:
x["block_type"] = (
"群聊"
if x["block_type"] == "group"
else "私聊"
if x["block_type"] == "private"
else "全部"
)
data["plugin_manager"] = PluginManager(**x)
if x := plugins2settings_manager.get(model):
if x.get("cmd") and isinstance(x.get("cmd"), list):
x["cmd"] = ",".join(x["cmd"])
data["plugin_settings"] = PluginSettings(**x)
if x := plugins2cd_manager.get(model):
data["cd_limit"] = CdLimit(**x)
if x := plugins2block_manager.get(model):
data["block_limit"] = BlockLimit(**x)
if x := plugins2count_manager.get(model):
data["count_limit"] = CountLimit(**x)
if x := Config.get(model):
id_ = 0
tmp = []
for key in x.keys():
tmp.append(
PluginConfig(
**{
"key": key,
"help_": x[key].get("help"),
"id": id_,
**x[key],
}
)
)
id_ += 1
data["plugin_config"] = tmp
plugin_list.append(Plugin(**data))
except Exception as e:
logger.error(
f"WEB_UI GET /webui/plugins model:{model} 发生错误 {type(e)}:{e}"
)
return Result(
code=500,
data=f"WEB_UI GET /webui/plugins model:{model} 发生错误 {type(e)}:{e}",
)
return Result(code=200, data=plugin_list)
@app.post("/webui/plugins")
def _(plugin: Plugin, user: User = Depends(token_to_user)) -> Result:
"""
修改插件信息
:param plugin: 插件内容
"""
try:
if plugin.plugin_config:
for c in plugin.plugin_config:
if str(c.value).lower() in ["true", "false"] and (
c.default_value is None or isinstance(c.default_value, bool)
):
c.value = True if str(c.value).lower() == "true" else False
elif isinstance(
Config.get_config(plugin.model, c.key, c.value), int
) or isinstance(c.default_value, int):
c.value = int(c.value)
elif isinstance(
Config.get_config(plugin.model, c.key, c.value), float
) or isinstance(c.default_value, float):
c.value = float(c.value)
elif isinstance(c.value, str) and (
isinstance(Config.get_config(plugin.model, c.key, c.value), list)
or isinstance(c.default_value, list)
):
c.value = c.value.split(",")
Config.set_config(plugin.model, c.key, c.value)
Config.save(None, True)
else:
if plugin.plugin_settings:
for key, value in plugin.plugin_settings:
plugins2settings_manager.set_module_data(plugin.model, key, value)
if plugin.plugin_manager:
for key, value in plugin.plugin_manager:
plugins_manager.set_module_data(plugin.model, key, value)
except Exception as e:
logger.error(
f"WEB_UI POST /webui/plugins model:{plugin.model} 发生错误 {type(e)}:{e}"
)
return Result(
code=500,
data=f"WEB_UI POST /webui/plugins model:{plugin.model} 发生错误 {type(e)}:{e}",
)
return Result(code=200)
+65
View File
@@ -0,0 +1,65 @@
from utils.manager import requests_manager
from ..auth import token_to_user, Depends, User
from utils.utils import get_bot
from models.group_info import GroupInfo
from ..config import *
@app.get("/webui/request")
def _(type_: Optional[str], user: User = Depends(token_to_user)) -> Result:
req_data = requests_manager.get_data()
req_list = []
if type_ in ["group", "private"]:
req_data = req_data[type_]
for x in req_data:
req_data[x]["oid"] = x
req_list.append(RequestResult(**req_data[x]))
req_list.reverse()
return Result(code=200, data=req_list)
@app.delete("/webui/request")
def _(type_: Optional[str], user: User = Depends(token_to_user)) -> Result:
"""
清空请求
:param type_: 类型
"""
requests_manager.clear(type_)
return Result(code=200)
@app.post("/webui/request")
async def _(parma: RequestParma, user: User = Depends(token_to_user)) -> Result:
"""
操作请求
:param parma: 参数
"""
result = "ok"
flag = 3
if bot := get_bot():
if parma.handle == "approve":
if parma.type == "group":
rid = requests_manager.get_group_id(parma.id)
if await GroupInfo.get_group_info(rid):
await GroupInfo.set_group_flag(rid, 1)
else:
group_info = await bot.get_group_info(group_id=rid)
await GroupInfo.add_group_info(
rid,
group_info["group_name"],
group_info["max_member_count"],
group_info["member_count"],
1,
)
flag = await requests_manager.approve(bot, parma.id, parma.type)
elif parma.handle == "refuse":
flag = await requests_manager.refused(bot, parma.id, parma.type)
elif parma.handle == "delete":
requests_manager.delete_request(parma.id, parma.type)
if parma.handle != "delete":
if flag == 1:
result = "该请求已失效"
requests_manager.delete_request(parma.id, parma.type)
elif flag == 2:
result = "未找到此Id"
return Result(code=200, data=result)
+219
View File
@@ -0,0 +1,219 @@
import asyncio
import os
from pathlib import Path
import psutil
import ujson as json
from configs.path_config import (
DATA_PATH,
FONT_PATH,
IMAGE_PATH,
LOG_PATH,
RECORD_PATH,
TEMP_PATH,
TEXT_PATH,
)
from services.log import logger
from utils.http_utils import AsyncHttpx
from ..auth import Depends, User, token_to_user
from ..config import *
CPU_DATA_PATH = DATA_PATH / "system" / "cpu.json"
MEMORY_DATA_PATH = DATA_PATH / "system" / "memory.json"
DISK_DATA_PATH = DATA_PATH / "system" / "disk.json"
CPU_DATA_PATH.parent.mkdir(exist_ok=True, parents=True)
cpu_data = {"data": []}
memory_data = {"data": []}
disk_data = {"data": []}
@app.get("/webui/system")
async def _(user: User = Depends(token_to_user)) -> Result:
return await get_system_data()
@app.get("/webui/system/status")
async def _(user: User = Depends(token_to_user)) -> Result:
return Result(
code=200,
data=await asyncio.get_event_loop().run_in_executor(None, _get_system_status),
)
@app.get("/webui/system/disk")
async def _(type_: Optional[str] = None, user: User = Depends(token_to_user)) -> Result:
return Result(
code=200,
data=await asyncio.get_event_loop().run_in_executor(
None, _get_system_disk, type_
),
)
@app.get("/webui/system/statusList")
async def _(user: User = Depends(token_to_user)) -> Result:
global cpu_data, memory_data, disk_data
await asyncio.get_event_loop().run_in_executor(None, _get_system_status)
cpu_rst = cpu_data["data"][-10:] if len(cpu_data["data"]) > 10 else cpu_data["data"]
memory_rst = (
memory_data["data"][-10:]
if len(memory_data["data"]) > 10
else memory_data["data"]
)
disk_rst = (
disk_data["data"][-10:] if len(disk_data["data"]) > 10 else disk_data["data"]
)
return Result(
code=200,
data=SystemStatusList(
cpu_data=cpu_rst,
memory_data=memory_rst,
disk_data=disk_rst,
),
)
async def get_system_data():
"""
说明:
获取系统信息,资源文件大小,网络状态等
"""
baidu = 200
google = 200
try:
await AsyncHttpx.get("https://www.baidu.com/", timeout=5)
except Exception as e:
logger.warning(f"访问BaiDu失败... {type(e)}: {e}")
baidu = 404
try:
await AsyncHttpx.get("https://www.google.com/", timeout=5)
except Exception as e:
logger.warning(f"访问Google失败... {type(e)}: {e}")
google = 404
network = SystemNetwork(baidu=baidu, google=google)
disk = await asyncio.get_event_loop().run_in_executor(None, _get_system_disk)
status = await asyncio.get_event_loop().run_in_executor(None, _get_system_status)
return Result(
code=200,
data=SystemResult(
status=status,
network=network,
disk=disk,
check_time=datetime.now().replace(microsecond=0),
),
)
def _get_system_status() -> SystemStatus:
"""
说明:
获取系统信息等
"""
cpu = psutil.cpu_percent()
memory = psutil.virtual_memory().percent
disk = psutil.disk_usage("/").percent
save_system_data(cpu, memory, disk)
return SystemStatus(
cpu=cpu,
memory=memory,
disk=disk,
check_time=datetime.now().replace(microsecond=0),
)
def _get_system_disk(
type_: Optional[str],
) -> Union[SystemFolderSize, Dict[str, Union[float, datetime]]]:
"""
说明:
获取资源文件大小等
"""
if not type_:
disk = SystemFolderSize(
font_dir_size=_get_dir_size(FONT_PATH) / 1024 / 1024,
image_dir_size=_get_dir_size(IMAGE_PATH) / 1024 / 1024,
text_dir_size=_get_dir_size(TEXT_PATH) / 1024 / 1024,
record_dir_size=_get_dir_size(RECORD_PATH) / 1024 / 1024,
temp_dir_size=_get_dir_size(TEMP_PATH) / 1024 / 102,
data_dir_size=_get_dir_size(DATA_PATH) / 1024 / 1024,
log_dir_size=_get_dir_size(LOG_PATH) / 1024 / 1024,
check_time=datetime.now().replace(microsecond=0),
)
return disk
else:
if type_ == "image":
dir_path = IMAGE_PATH
elif type_ == "font":
dir_path = FONT_PATH
elif type_ == "text":
dir_path = TEXT_PATH
elif type_ == "record":
dir_path = RECORD_PATH
elif type_ == "data":
dir_path = DATA_PATH
elif type_ == "temp":
dir_path = TEMP_PATH
else:
dir_path = LOG_PATH
dir_map = {}
other_file_size = 0
for file in os.listdir(dir_path):
file = Path(dir_path / file)
if file.is_dir():
dir_map[file.name] = _get_dir_size(file) / 1024 / 1024
else:
other_file_size += os.path.getsize(file) / 1024 / 1024
dir_map["其他文件"] = other_file_size
dir_map["check_time"] = datetime.now().replace(microsecond=0)
return dir_map
def _get_dir_size(dir_path: Path) -> float:
"""
说明:
获取文件夹大小
参数:
:param dir_path: 文件夹路径
"""
size = 0
for root, dirs, files in os.walk(dir_path):
size += sum([os.path.getsize(os.path.join(root, name)) for name in files])
return size
def save_system_data(cpu: float, memory: float, disk: float):
"""
说明:
保存一些系统信息
参数:
:param cpu: cpu
:param memory: memory
:param disk: disk
"""
global cpu_data, memory_data, disk_data
if CPU_DATA_PATH.exists() and not cpu_data["data"]:
with open(CPU_DATA_PATH, "r") as f:
cpu_data = json.load(f)
if MEMORY_DATA_PATH.exists() and not memory_data["data"]:
with open(MEMORY_DATA_PATH, "r") as f:
memory_data = json.load(f)
if DISK_DATA_PATH.exists() and not disk_data["data"]:
with open(DISK_DATA_PATH, "r") as f:
disk_data = json.load(f)
now = str(datetime.now().time().replace(microsecond=0))
cpu_data["data"].append({"time": now, "data": cpu})
memory_data["data"].append({"time": now, "data": memory})
disk_data["data"].append({"time": now, "data": disk})
if len(cpu_data["data"]) > 50:
cpu_data["data"] = cpu_data["data"][-50:]
if len(memory_data["data"]) > 50:
memory_data["data"] = memory_data["data"][-50:]
if len(disk_data["data"]) > 50:
disk_data["data"] = disk_data["data"][-50:]
with open(CPU_DATA_PATH, "w") as f:
json.dump(cpu_data, f, indent=4, ensure_ascii=False)
with open(MEMORY_DATA_PATH, "w") as f:
json.dump(memory_data, f, indent=4, ensure_ascii=False)
with open(DISK_DATA_PATH, "w") as f:
json.dump(disk_data, f, indent=4, ensure_ascii=False)
+101 -1
View File
@@ -1,6 +1,7 @@
from typing import Optional, List, Any, Union
from typing import Optional, List, Any, Union, Dict
from pydantic import BaseModel
from fastapi.middleware.cors import CORSMiddleware
from datetime import datetime
import nonebot
@@ -18,6 +19,9 @@ app.add_middleware(
class CdLimit(BaseModel):
"""
Cd 限制
"""
cd: int
status: bool
check_type: str
@@ -26,6 +30,9 @@ class CdLimit(BaseModel):
class BlockLimit(BaseModel):
"""
Block限制
"""
status: bool
check_type: str
limit_type: str
@@ -33,6 +40,9 @@ class BlockLimit(BaseModel):
class CountLimit(BaseModel):
"""
Count限制
"""
max_count: int
status: bool
limit_type: bool
@@ -40,6 +50,9 @@ class CountLimit(BaseModel):
class PluginManager(BaseModel):
"""
插件信息
"""
plugin_name: str # 插件名称
status: Optional[bool] # 插件状态
error: Optional[bool] # 加载状态
@@ -49,6 +62,9 @@ class PluginManager(BaseModel):
class PluginSettings(BaseModel):
"""
插件基本设置
"""
level: Optional[int] # 群权限等级
default_status: Optional[bool] # 默认开关
limit_superuser: Optional[bool] # 是否限制超级用户
@@ -57,16 +73,34 @@ class PluginSettings(BaseModel):
plugin_type: Optional[List[Union[str, int]]] # 帮助类型
class PluginConfig(BaseModel):
"""
插件配置项
"""
id: int
key: str
value: Optional[Any]
help_: Optional[str]
default_value: Optional[Any]
class Plugin(BaseModel):
"""
插件
"""
model: str # 模块
plugin_settings: Optional[PluginSettings]
plugin_manager: Optional[PluginManager]
plugin_config: Optional[List[PluginConfig]]
cd_limit: Optional[CdLimit]
block_limit: Optional[BlockLimit]
count_limit: Optional[CountLimit]
class Group(BaseModel):
"""
群组信息
"""
group_id: int
group_name: str
member_count: int
@@ -74,12 +108,18 @@ class Group(BaseModel):
class Task(BaseModel):
"""
被动技能
"""
name: str
nameZh: str
status: bool
class GroupResult(BaseModel):
"""
群组返回数据
"""
group: Group
level: int
status: bool
@@ -88,6 +128,9 @@ class GroupResult(BaseModel):
class RequestResult(BaseModel):
"""
好友/群组请求管理
"""
oid: str
id: int
flag: str
@@ -102,11 +145,68 @@ class RequestResult(BaseModel):
class RequestParma(BaseModel):
"""
操作请求接收数据
"""
id: int
handle: str
type: str
class SystemStatus(BaseModel):
"""
系统状态
"""
cpu: int
memory: int
disk: int
check_time: datetime
class SystemNetwork(BaseModel):
"""
系统网络状态
"""
baidu: int
google: int
class SystemFolderSize(BaseModel):
"""
资源文件占比
"""
font_dir_size: float
image_dir_size: float
text_dir_size: float
record_dir_size: float
temp_dir_size: float
data_dir_size: float
log_dir_size: float
check_time: datetime
class SystemStatusList(BaseModel):
"""
状态记录
"""
cpu_data: List[Dict[str, Union[float, str]]]
memory_data: List[Dict[str, Union[float, str]]]
disk_data: List[Dict[str, Union[float, str]]]
class SystemResult(BaseModel):
"""
系统api返回
"""
status: SystemStatus
network: SystemNetwork
disk: SystemFolderSize
check_time: datetime
class Result(BaseModel):
"""
总体返回
"""
code: int
data: Any
+2 -2
View File
@@ -28,13 +28,13 @@ class WordBankBuilder:
"""
self._data["answer"] = answer
async def save(self):
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")
await WordBank.add_problem_answer(user_id, group_id, 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"]
+33 -23
View File
@@ -25,6 +25,7 @@ class WordBank(db.Model):
cls,
user_id: int,
group_id: Optional[int],
search_type: [int],
problem: str,
answer: str,
format_: Optional[List[Tuple[int, Union[int, str]]]],
@@ -33,6 +34,7 @@ class WordBank(db.Model):
添加或新增一个问答
:param user_id: 用户id
:param group_id: 群号
:search_type: 问题类型,
:param problem: 问题
:param answer: 回答
:param format_: 格式化数据
@@ -43,7 +45,7 @@ class WordBank(db.Model):
for x, y in format_:
_str += f"{x}<_s>{y}<format>"
return await cls._problem_answer_handle(
user_id, group_id, problem, "add", answer=answer, format_=_str
user_id, group_id, problem, "add", search_type=search_type, answer=answer, format_=_str
)
@classmethod
@@ -159,22 +161,25 @@ class WordBank(db.Model):
if problem:
FUZZY = Config.get_config("word_bank", "WORD_BANK_FUZZY")
KEY = Config.get_config("word_bank", "WORD_BANK_KEY")
q = await cls.query.where(
(cls.group_id == group_id) & (cls.problem == problem)
).gino.all()
if KEY and FUZZY:
q_fuzzy = await cls.query.where(
(cls.group_id == group_id) & (cls.problem.contains(f'{problem}'))).gino.all()
q_key = await cls.query.where(cls.group_id == group_id).gino.all()
(cls.group_id == group_id) & (cls.search_type == 2) & (
cls.problem.contains(f'{problem}'))).gino.all()
q_key = await cls.query.where((cls.group_id == group_id) & (cls.search_type == 1)).gino.all()
q_key = [x for x in q_key if str(x.problem) in (problem)]
q = q_fuzzy + q_key
q += q_fuzzy + q_key
elif FUZZY:
q = await cls.query.where(
(cls.group_id == group_id) & (cls.problem.contains(f'{problem}'))).gino.all()
q_fuzzy = await cls.query.where(
(cls.group_id == group_id) & (cls.search_type == 2) & (
cls.problem.contains(f'{problem}'))).gino.all()
q += q_fuzzy
elif KEY:
q = await cls.query.where(cls.group_id == group_id).gino.all()
q = [x for x in q if str(x.problem) in (problem)]
else:
q = await cls.query.where(
(cls.group_id == group_id) & (cls.problem == problem)
).gino.all()
q_key = await cls.query.where((cls.group_id == group_id) & (cls.search_type == 1)).gino.all()
q_key = [x for x in q_key if str(x.problem) in (problem)]
q += q_key
else:
return None
@@ -188,6 +193,7 @@ class WordBank(db.Model):
problem: str,
type_: str,
*,
search_type: [int] = 0,
answer: Optional[str] = None,
index: Optional[int] = None,
format_: Optional[str] = None,
@@ -213,17 +219,21 @@ class WordBank(db.Model):
else:
q = cls.query.where((cls.user_qq == user_id) & (cls.problem == problem))
if type_ == "add":
q = await q.where(cls.answer == answer).gino.all()
if not q or ".jpg" in format_:
await cls.create(
user_qq=user_id,
group_id=group_id,
problem=problem,
answer=answer,
format=format_,
create_time=datetime.now().replace(microsecond=0),
update_time=datetime.now().replace(microsecond=0),
)
q = await q.where((cls.answer == answer) & (cls.search_type == search_type)).gino.all()
try:
if not q or ".jpg" in format_:
await cls.create(
user_qq=user_id,
group_id=group_id,
search_type=search_type,
problem=problem,
answer=answer,
format=format_,
create_time=datetime.now().replace(microsecond=0),
update_time=datetime.now().replace(microsecond=0),
)
except:
return False
return True
elif type_ == "delete":
q = await q.with_for_update().gino.all()
+15 -9
View File
@@ -4,7 +4,6 @@ from services.log import logger
from configs.path_config import DATA_PATH
from utils.http_utils import AsyncHttpx
from ._data_source import WordBankBuilder
from configs.config import Config
from utils.message_builder import image
from utils.image_utils import text2image
from .message_handle import get_one_answer
@@ -27,7 +26,7 @@ usage:
对指定问题的随机回答,对相同问题可以设置多个不同回答
删除词条后每个词条的id可能会变化,请查看后再删除
指令:
添加词条问...答...:添加问答词条,可重复添加相同问题的不同回答
添加词条 ?[模糊/关键字]...答...:添加问答词条,可重复添加相同问题的不同回答
删除词条 [问题/下标] ?[下标]:删除指定词条指定或全部回答
修改词条 [问题/下标] ?[下标/新回答] [新回答]:修改指定词条指定回答默认修改为第一条
查看词条 ?[问题/下标]:查看全部词条或对应词条回答
@@ -44,13 +43,13 @@ usage:
""".strip()
__plugin_des__ = "自定义词条内容随机回复"
__plugin_cmd__ = [
"添加词条问...答..",
"添加词条 ?[模糊/关键字]问...答..",
"删除词条 [问题/下标] ?[下标]",
"修改词条 [问题/下标] ?[下标/新回答] [新回答]",
"查看词条 ?[问题/下标]",
]
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_version__ = 0.2
__plugin_author__ = "HibiKier & yajiwa"
__plugin_settings__ = {
"admin_level": Config.get_config("word_bank", "WORD_BANK_LEVEL [LEVEL]"),
"cmd": ["词库问答", "添加词条", "删除词条", "修改词条", "查看词条"],
@@ -71,7 +70,7 @@ show_word = on_command("显示词条", aliases={"查看词条"}, priority=5, blo
@add_word.handle()
async def _(bot: Bot, event: GroupMessageEvent, state: T_State, arg: Message = CommandArg()):
msg = str(arg)
r = re.search(r"^问(.+)\s?答([\s\S]*)", msg)
r = re.search(r"问(.+)\s?答([\s\S]*)", msg)
if not r:
await add_word.finish("未检测到词条问题...")
problem = r.group(1).strip()
@@ -87,10 +86,17 @@ async def _(bot: Bot, event: GroupMessageEvent, state: T_State, arg: Message = C
break
else:
_problem = problem
search_type = 0
if re.search("^关键字(.*)", msg):
search_type = 1
elif re.search("^模糊(.*)", msg):
search_type = 2
_builder = await get__builder(event, _problem, answer, idx)
await _builder.save()
logger.info(f"已保存词条 问:{problem} 答:{msg}")
await add_word.send(f"已保存词条:{problem}")
if await _builder.save(search_type):
logger.info(f"已保存词条 问:{problem} 答:{msg}")
await add_word.send(f"已保存词条:{problem}")
else:
await delete_word.send("保存失败,可能是回答重复")
@delete_word.handle()
+8 -8
View File
@@ -11,7 +11,7 @@ china_city = TEXT_PATH / "china_city.json"
data = {}
url = "https://view.inews.qq.com/g2/getOnsInfo?name=disease_h5"
url = "https://api.inews.qq.com/newsqa/v1/query/inner/publish/modules/list?modules=diseaseh5Shelf"
async def get_yiqing_data(area: str) -> Union[str, MessageSegment]:
@@ -37,7 +37,7 @@ async def get_yiqing_data(area: str) -> Union[str, MessageSegment]:
if area in data[p]:
province = p
city = area
epidemic_data = json.loads((await AsyncHttpx.get(url)).json()["data"])
epidemic_data = (await AsyncHttpx.get(url)).json()["data"]["diseaseh5Shelf"]
last_update_time = epidemic_data["lastUpdateTime"]
if area == "中国":
data_ = epidemic_data["areaTree"][0]
@@ -59,12 +59,12 @@ async def get_yiqing_data(area: str) -> Union[str, MessageSegment]:
add_confirm = data_["today"]["confirm"] # 新增确诊
grade = ""
_grade_color = ""
if data_["total"].get("grade"):
grade = data_["total"]["grade"]
if "中风险" in grade:
_grade_color = "#fa9424"
else:
_grade_color = "red"
# if data_["total"].get("grade"):
# grade = data_["total"]["grade"]
# if "中风险" in grade:
# _grade_color = "#fa9424"
# else:
# _grade_color = "red"
dead_rate = f"{dead / confirm * 100:.2f}" # 死亡率
heal_rate = f"{heal / confirm * 100:.2f}" # 治愈率
+6 -2
View File
@@ -14,16 +14,20 @@ _browser: Optional[Browser] = None
async def init(**kwargs) -> Optional[Browser]:
global _browser
if platform.system() == "Windows":
return None
try:
global _browser
browser = await async_playwright().start()
_browser = await browser.chromium.launch(**kwargs)
return _browser
except NotImplementedError:
logger.warning("win环境下 初始化playwright失败,相关功能将被限制....")
return None
except Exception as e:
logger.warning(f"启动chromium发生错误 {type(e)}:{e}")
if _browser:
await _browser.close()
return None
async def get_browser(**kwargs) -> Browser:
View File
+110
View File
@@ -0,0 +1,110 @@
from typing import Callable, Union, Tuple
from nonebot.plugin import require
use = require("use")
shop = require("shop_handle")
class ShopRegister(dict):
def __init__(self, *args, **kwargs):
super(ShopRegister, self).__init__(*args, **kwargs)
self._data = {}
self._flag = True
def register(
self,
name: Tuple[str, ...],
price: Tuple[float, ...],
des: Tuple[str, ...],
load_status: Tuple[bool, ...],
**kwargs,
):
def add_register_item(func: Callable):
if name in self._data.keys():
raise ValueError("该商品已注册,请替换其他名称!")
for n, p, d, s in zip(name, price, des, load_status):
if s:
_temp_kwargs = {}
for key, value in kwargs.items():
if key.startswith(f"{n}_"):
_temp_kwargs[key.split("_", maxsplit=1)[-1]] = value
self._data[n] = {
"price": p,
"des": d,
"func": func,
"kwargs": _temp_kwargs,
}
return func
return lambda func: add_register_item(func)
async def load_register(self):
# 统一进行注册
if self._flag:
# 只进行一次注册
self._flag = False
for name in self._data.keys():
await shop.register_goods(
name, self._data[name]["price"], self._data[name]["des"]
)
use.register_use(
name, self._data[name]["func"], **self._data[name]["kwargs"]
)
def __call__(
self,
name: Union[str, Tuple[str, ...]],
price: Union[float, Tuple[float, ...]],
des: Union[str, Tuple[str, ...]],
load_status: Union[bool, Tuple[bool, ...]] = True,
**kwargs,
):
_tuple_list = []
_current_len = -1
for x in [name, price, des, load_status]:
if isinstance(x, tuple):
if _current_len == -1:
_current_len = len(x)
if _current_len != len(x):
raise ValueError(f"注册商品 {name} 中 name,price,des,load_status 数量不符!")
_current_len = _current_len if _current_len > -1 else 1
_name = name if isinstance(name, tuple) else tuple(name)
_price = (
price
if isinstance(price, tuple)
else tuple([price for _ in range(_current_len)])
)
_des = (
des if isinstance(des, tuple) else tuple([des for _ in range(_current_len)])
)
_load_status = (
load_status
if isinstance(load_status, tuple)
else tuple([load_status for _ in range(_current_len)])
)
return self.register(_name, _price, _des, _load_status, **kwargs)
def __setitem__(self, key, value):
self._data[key] = value
def __getitem__(self, key):
return self._data[key]
def __contains__(self, key):
return key in self._data
def __str__(self):
return str(self._data)
def keys(self):
return self._data.keys()
def values(self):
return self._data.values()
def items(self):
return self._data.items()
shop_register = ShopRegister()
View File
+23 -19
View File
@@ -1,17 +1,18 @@
import asyncio
from configs.path_config import IMAGE_PATH, FONT_PATH
from PIL import Image, ImageFile, ImageDraw, ImageFont, ImageFilter
from imagehash import ImageHash
from io import BytesIO
from matplotlib import pyplot as plt
from typing import Tuple, Optional, Union, List, Literal
from pathlib import Path
from math import ceil
import random
import cv2
import base64
import imagehash
import random
import re
from io import BytesIO
from math import ceil
from pathlib import Path
from typing import List, Literal, Optional, Tuple, Union
import cv2
import imagehash
from configs.path_config import FONT_PATH, IMAGE_PATH
from imagehash import ImageHash
from matplotlib import pyplot as plt
from PIL import Image, ImageDraw, ImageFile, ImageFilter, ImageFont
ImageFile.LOAD_TRUNCATED_IMAGES = True
Image.MAX_IMAGE_PIXELS = None
@@ -65,9 +66,7 @@ def compressed_image(
"""
in_file = IMAGE_PATH / in_file if isinstance(in_file, str) else in_file
if out_file:
out_file = (
IMAGE_PATH / out_file if isinstance(out_file, str) else out_file
)
out_file = IMAGE_PATH / out_file if isinstance(out_file, str) else out_file
else:
out_file = in_file
h, w, d = cv2.imread(str(in_file.absolute())).shape
@@ -673,9 +672,11 @@ class BuildImage:
ellipse_box = [0, 0, r2 - 2, r2 - 2]
mask = Image.new(
size=[int(dim * antialias) for dim in self.markImg.size],
mode='L', color='black')
mode="L",
color="black",
)
draw = ImageDraw.Draw(mask)
for offset, fill in (width / -2.0, 'black'), (width / 2.0, 'white'):
for offset, fill in (width / -2.0, "black"), (width / 2.0, "white"):
left, top = [(value + offset) * antialias for value in ellipse_box[:2]]
right, bottom = [(value - offset) * antialias for value in ellipse_box[2:]]
draw.ellipse([left, top, right, bottom], fill=fill)
@@ -1335,6 +1336,7 @@ async def text2image(
font: str = "CJGaoDeGuo.otf",
font_color: Union[str, Tuple[int, int, int]] = "black",
padding: Union[int, Tuple[int, int, int, int]] = 0,
_add_height: float = 0,
) -> BuildImage:
"""
说明:
@@ -1357,6 +1359,7 @@ async def text2image(
:param font: 普通字体
:param font_color: 普通字体颜色
:param padding: 文本外边距,元组类型时为 (上,左,下,右)
:param _add_height: 由于get_size无法返回正确的高度,采用手动方式额外添加高度
"""
pw = ph = top_padding = left_padding = 0
if padding:
@@ -1488,16 +1491,17 @@ async def text2image(
else:
width = 0
height = 0
_tmp = BuildImage(0, 0, font_size=font_size)
_tmp = BuildImage(0, 0, font=font, font_size=font_size)
for x in text.split("\n"):
x = x if x.strip() else "正"
w, h = _tmp.getsize(x)
height += h
height += h + _add_height
width = width if width > w else w
width += pw
height += ph
A = BuildImage(
width + left_padding,
height + top_padding,
height + top_padding + 2,
font_size=font_size,
color=color,
font=font,
+9 -8
View File
@@ -85,7 +85,7 @@ class RequestManager(StaticData):
return data["invite_group"]
return None
async def approve(self, bot: Bot, id_: int, type_: str) -> Optional[int]:
async def approve(self, bot: Bot, id_: int, type_: str) -> int:
"""
同意请求
:param bot: Bot
@@ -232,7 +232,7 @@ class RequestManager(StaticData):
async def _set_add_request(
self, bot: Bot, id_: int, type_: str, approve: bool
) -> Optional[int]:
) -> int:
"""
处理请求
:param bot: Bot
@@ -260,12 +260,13 @@ class RequestManager(StaticData):
f"同意{self._data[type_][id_]['nickname']}({self._data[type_][id_]['id']})"
f"的{'好友' if type_ == 'private' else '入群'}请求失败了..."
)
return None
logger.info(
f"{'同意' if approve else '拒绝'}{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 None
return 2 # 未找到id
+46 -12
View File
@@ -1,16 +1,16 @@
from configs.path_config import IMAGE_PATH, RECORD_PATH
from nonebot.adapters.onebot.v11.message import MessageSegment
from configs.config import NICKNAME
from services.log import logger
from typing import Union, List
from pathlib import Path
import os
from typing import List, Union
from configs.config import NICKNAME
from configs.path_config import IMAGE_PATH, RECORD_PATH
from nonebot.adapters.onebot.v11.message import MessageSegment, Message
from services.log import logger
def image(
file: Union[str, Path, bytes] = None,
path: str = None,
b64: str = None,
file: Union[str, Path, bytes] = None,
path: str = None,
b64: str = None,
) -> Union[MessageSegment, str]:
"""
说明:
@@ -63,7 +63,9 @@ def record(voice_name: str, path: str = None) -> MessageSegment or str:
if len(voice_name.split(".")) == 1:
voice_name += ".mp3"
file = (
Path(RECORD_PATH) / path / voice_name if path else Path(RECORD_PATH) / voice_name
Path(RECORD_PATH) / path / voice_name
if path
else Path(RECORD_PATH) / voice_name
)
if "http" in voice_name:
return MessageSegment.record(voice_name)
@@ -96,7 +98,7 @@ def contact_user(qq: int) -> MessageSegment:
def share(
url: str, title: str, content: str = None, image_url: str = None
url: str, title: str, content: str = None, image_url: str = None
) -> MessageSegment:
"""
说明:
@@ -155,7 +157,7 @@ def music(type_: str, id_: int) -> MessageSegment:
def custom_forward_msg(
msg_list: List[str], uin: Union[int, str], name: str = f"这里是{NICKNAME}"
msg_list: List[str], uin: Union[int, str], name: str = f"这里是{NICKNAME}"
) -> List[dict]:
"""
生成自定义合并消息
@@ -176,3 +178,35 @@ def custom_forward_msg(
}
mes_list.append(data)
return mes_list
class MessageBuilder:
"""
MessageSegment构建工具
"""
def __init__(self, msg: Union[str, MessageSegment, Message]):
if msg:
if isinstance(msg, str):
self._msg = text(msg)
else:
self._msg = msg
else:
self._msg = text("")
def text(self, msg: str):
return MessageBuilder(self._msg + text(msg))
def image(
self,
file: Union[str, Path, bytes] = None,
path: str = None,
b64: str = None,
):
return MessageBuilder(self._msg + image(file, path, b64))
def at(self, qq: int):
return MessageBuilder(self._msg + at(qq))
def face(self, id_: int):
return MessageBuilder(self._msg + face(id_))
+14
View File
@@ -0,0 +1,14 @@
from nonebot.adapters.onebot.v11 import Bot, MessageEvent
from pydantic import BaseModel, create_model
from typing import Any
class ShopParam(BaseModel):
goods_name: str
user_id: int
group_id: int
bot: Any
event: MessageEvent
num: int # 道具单次使用数量
send_success_msg: bool = True # 是否发送使用成功信息
max_num_limit: int = 1 # 单次使用最大次数