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

🐛 bug修复

Browse Source
This commit is contained in:
HibiKier 2024-09-01 23:30:26 +08:00
parent 7e7436f433
commit 8fe061738a
6 changed files with 90 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
requirements.txt
View File

Binary file not shown.

24
zhenxun/builtin_plugins/shop/__init__.py
View File

@ -1,22 +1,22 @@
from nonebot.adapters import Bot, Event
from nonebot.plugin import PluginMetadata
from nonebot_plugin_session import EventSession
from nonebot_plugin_userinfo import UserInfo, EventUserInfo
from nonebot_plugin_alconna import (
Alconna,
Args,
UniMsg,
Alconna,
Arparma,
Subcommand,
UniMessage,
UniMsg,
on_alconna,
)
from nonebot_plugin_session import EventSession
from nonebot_plugin_userinfo import EventUserInfo, UserInfo
from zhenxun.configs.utils import BaseBlock, PluginExtraData
from zhenxun.services.log import logger
from zhenxun.utils.enum import BlockType, PluginType
from zhenxun.utils.exception import GoodsNotFound
from zhenxun.utils.message import MessageUtils
from zhenxun.utils.exception import GoodsNotFound
from zhenxun.utils.enum import BlockType, PluginType
from zhenxun.configs.utils import BaseBlock, PluginExtraData
from ._data_source import ShopManage
@ -96,7 +96,7 @@ async def _(session: EventSession, arparma: Arparma):
gold = await ShopManage.my_cost(session.id1, session.platform)
await MessageUtils.build_message(f"你的当前余额: {gold}").send(reply_to=True)
else:
await MessageUtils.build_message(f"用户id为空...").send(reply_to=True)
await MessageUtils.build_message("用户id为空...").send(reply_to=True)
@_matcher.assign("my-props")
@ -111,11 +111,9 @@ async def _(
session.platform,
):
await MessageUtils.build_message(image.pic2bytes()).finish(reply_to=True)
return await MessageUtils.build_message(f"你的道具为空捏...").send(
reply_to=True
)
return await MessageUtils.build_message("你的道具为空捏...").send(reply_to=True)
else:
await MessageUtils.build_message(f"用户id为空...").send(reply_to=True)
await MessageUtils.build_message("用户id为空...").send(reply_to=True)
@_matcher.assign("buy")
@ -129,7 +127,7 @@ async def _(session: EventSession, arparma: Arparma, name: str, num: int):
result = await ShopManage.buy_prop(session.id1, name, num, session.platform)
await MessageUtils.build_message(result).send(reply_to=True)
else:
await MessageUtils.build_message(f"用户id为空...").send(reply_to=True)
await MessageUtils.build_message("用户id为空...").send(reply_to=True)
@_matcher.assign("use")

97
zhenxun/builtin_plugins/shop/_data_source.py
View File

@ -1,21 +1,22 @@
import time
import asyncio
import inspect
import time
from typing import Any, Literal
from types import MappingProxyType
from typing import Any, Callable, Literal
from collections.abc import Callable
from nonebot.adapters import Bot, Event
from nonebot_plugin_alconna import UniMessage, UniMsg
from nonebot_plugin_session import EventSession
from pydantic import BaseModel, create_model
from nonebot_plugin_session import EventSession
from nonebot_plugin_alconna import UniMsg, UniMessage
from zhenxun.configs.path_config import IMAGE_PATH
from zhenxun.services.log import logger
from zhenxun.models.goods_info import GoodsInfo
from zhenxun.configs.path_config import IMAGE_PATH
from zhenxun.models.user_console import UserConsole
from zhenxun.models.user_gold_log import UserGoldLog
from zhenxun.models.user_props_log import UserPropsLog
from zhenxun.services.log import logger
from zhenxun.utils.enum import GoldHandle, PropHandle
from zhenxun.models.user_props_log import UserPropsLog
from zhenxun.utils.image_utils import BuildImage, ImageTemplate, text2image
ICON_PATH = IMAGE_PATH / "shop_icon"
@ -65,11 +66,13 @@ class ShopParam(BaseModel):
"""单次使用最大次数"""
session: EventSession | None = None
"""EventSession"""
message: UniMsg
"""UniMessage"""
class ShopManage:
uuid2goods: dict[str, Goods] = {}
uuid2goods: dict[str, Goods] = {} # noqa: RUF012
@classmethod
def __build_params(
@ -102,6 +105,7 @@ class ShopManage:
"num": num,
"text": text,
"session": session,
"message": message,
}
)
return model, {
@ -113,6 +117,7 @@ class ShopManage:
"num": num,
"text": text,
"goods_name": goods.name,
"message": message,
}
@classmethod
@ -121,7 +126,6 @@ class ShopManage:
args: MappingProxyType,
param: ShopParam,
session: EventSession,
message: UniMsg,
**kwargs,
) -> list[Any]:
"""解析参数
@ -144,7 +148,7 @@ class ShopManage:
elif par in ["session"]:
param_list.append(session)
elif par in ["message"]:
param_list.append(message)
param_list.append(kwargs.get("message"))
elif par not in ["args", "kwargs"]:
param_list.append(param_json.get(par))
if kwargs.get(par) is not None:
@ -170,16 +174,15 @@ class ShopManage:
if fun_list:
for func in fun_list:
args = inspect.signature(func).parameters
if args and list(args.keys())[0] != "kwargs":
if args and next(iter(args.keys())) != "kwargs":
if asyncio.iscoroutinefunction(func):
await func(*cls.__parse_args(args, param, **kwargs))
else:
func(*cls.__parse_args(args, param, **kwargs))
elif asyncio.iscoroutinefunction(func):
await func(**kwargs)
else:
if asyncio.iscoroutinefunction(func):
await func(**kwargs)
else:
func(**kwargs)
func(**kwargs)
@classmethod
async def __run(
@ -187,7 +190,6 @@ class ShopManage:
goods: Goods,
param: ShopParam,
session: EventSession,
message: UniMsg,
**kwargs,
) -> str | UniMessage | None:
"""运行道具函数
@ -201,22 +203,18 @@ class ShopManage:
"""
args = inspect.signature(goods.func).parameters # type: ignore
if goods.func:
if args and list(args.keys())[0] != "kwargs":
if asyncio.iscoroutinefunction(goods.func):
return await goods.func(
*cls.__parse_args(args, param, session, message, **kwargs)
)
else:
return goods.func(
*cls.__parse_args(args, param, session, message, **kwargs)
)
if args and next(iter(args.keys())) != "kwargs":
return (
await goods.func(*cls.__parse_args(args, param, session, **kwargs))
if asyncio.iscoroutinefunction(goods.func)
else goods.func(*cls.__parse_args(args, param, session, **kwargs))
)
if asyncio.iscoroutinefunction(goods.func):
return await goods.func(
**kwargs,
)
else:
if asyncio.iscoroutinefunction(goods.func):
return await goods.func(
**kwargs,
)
else:
return goods.func(**kwargs)
return goods.func(**kwargs)
@classmethod
async def use(
@ -252,17 +250,17 @@ class ShopManage:
if not goods_info:
return f"{goods_name} 不存在..."
if goods_info.is_passive:
return f"{goods_name} 是被动道具, 无法使用..."
return f"{goods_info.goods_name} 是被动道具, 无法使用..."
goods = cls.uuid2goods.get(goods_info.uuid)
if not goods or not goods.func:
return f"{goods_name} 未注册使用函数, 无法使用..."
return f"{goods_info.goods_name} 未注册使用函数, 无法使用..."
param, kwargs = cls.__build_params(
bot, event, session, message, goods, num, text
)
if num > param.max_num_limit:
return f"{goods_info.goods_name} 单次使用最大数量为{param.max_num_limit}..."
await cls.run_before_after(goods, param, "before", **kwargs)
result = await cls.__run(goods, param, session, message, **kwargs)
result = await cls.__run(goods, param, session, **kwargs)
await UserConsole.use_props(session.id1, goods_info.uuid, num, session.platform) # type: ignore
await cls.run_before_after(goods, param, "after", **kwargs)
if not result and param.send_success_msg:
@ -334,11 +332,10 @@ class ShopManage:
]
if name.isdigit():
goods = goods_list[int(name) - 1]
elif filter_goods := [g for g in goods_list if g.goods_name == name]:
goods = filter_goods[0]
else:
if filter_goods := [g for g in goods_list if g.goods_name == name]:
goods = filter_goods[0]
else:
return "道具名称不存在..."
return "道具名称不存在..."
user = await UserConsole.get_user(user_id, platform)
price = goods.goods_price * num * goods.goods_discount
if user.gold < price:
@ -423,13 +420,12 @@ class ShopManage:
BuildImage: 商店图片
"""
goods_lst = await GoodsInfo.get_all_goods()
_dc = {}
font_h = BuildImage.get_text_size("正")[1]
h = 10
_list: list[GoodsInfo] = []
for goods in goods_lst:
if goods.goods_limit_time == 0 or time.time() < goods.goods_limit_time:
_list.append(goods)
_list: list[GoodsInfo] = [
goods
for goods in goods_lst
if goods.goods_limit_time == 0 or time.time() < goods.goods_limit_time
]
# A = BuildImage(1100, h, color="#f9f6f2")
total_n = 0
image_list = []
@ -471,7 +467,7 @@ class ShopManage:
440 + _tmp.width,
0,
),
f" 金币",
" 金币",
center_type="height",
)
des_image = None
@ -541,7 +537,7 @@ class ShopManage:
).split()
y_m_d = limit_time[0]
_h_m = limit_time[1].split(":")
h_m = _h_m[0] + "时 " + _h_m[1] + "分"
h_m = f"{_h_m[0]}时 {_h_m[1]}分"
await bk.text((_w + 55, 38), str(y_m_d))
await bk.text((_w + 65, 57), str(h_m))
_w += 140
@ -575,8 +571,7 @@ class ShopManage:
f"{goods.daily_limit}", size=30
)
await bk.paste(_tmp, (_w + 72, 45))
if total_n < n:
total_n = n
total_n = max(total_n, n)
if n:
await bk.line((650, -1, 650 + n, -1), "#a29ad6", 5)
# await bk.aline((650, 80, 650 + n, 80), "#a29ad6", 5)
@ -585,10 +580,8 @@ class ShopManage:
image_list.append(bk)
# await A.apaste(bk, (0, current_h), True)
# current_h += 90
h = 0
current_h = 0
for img in image_list:
h += img.height + 10
h = sum(img.height + 10 for img in image_list)
A = BuildImage(1100, h, color="#f9f6f2")
for img in image_list:
await A.paste(img, (0, current_h))
@ -597,7 +590,7 @@ class ShopManage:
if total_n:
w += total_n
h = A.height + 230 + 100
h = 1000 if h < 1000 else h
h = max(h, 1000)
shop_logo = BuildImage(100, 100, background=f"{IMAGE_PATH}/other/shop_text.png")
shop = BuildImage(w, h, font_size=20, color="#f9f6f2")
await shop.paste(A, (20, 230))

1
zhenxun/builtin_plugins/sign_in/__init__.py
View File

@ -17,6 +17,7 @@ from zhenxun.configs.utils import PluginCdBlock, RegisterConfig, PluginExtraData
from ._data_source import SignManage
from .utils import clear_sign_data_pic
from .goods_register import driver # noqa: F401
__plugin_meta__ = PluginMetadata(
name="签到",

24
zhenxun/builtin_plugins/sign_in/goods_register.py
View File

@ -6,7 +6,7 @@ from nonebot_plugin_session import EventSession
from zhenxun.models.sign_user import SignUser
from zhenxun.models.user_console import UserConsole
from zhenxun.utils.decorator.shop import NotMeetUseConditionsException, shop_register
from zhenxun.utils.decorator.shop import shop_register
driver: Driver = nonebot.get_driver()
@ -32,9 +32,13 @@ driver: Driver = nonebot.get_driver()
"favorability_card_2.png",
"favorability_card_3.png",
),
**{"好感度双倍加持卡Ⅰ_prob": 0.1, "好感度双倍加持卡Ⅱ_prob": 0.2, "好感度双倍加持卡Ⅲ_prob": 0.3}, # type: ignore
**{
"好感度双倍加持卡Ⅰ_prob": 0.1,
"好感度双倍加持卡Ⅱ_prob": 0.2,
"好感度双倍加持卡Ⅲ_prob": 0.3,
}, # type: ignore
)
async def _(session: EventSession, user_id: int, group_id: int, prob: float):
async def _(session: EventSession, user_id: int, prob: float):
if session.id1:
user_console = await UserConsole.get_user(session.id1, session.platform)
user, _ = await SignUser.get_or_create(
@ -53,20 +57,24 @@ async def _(session: EventSession, user_id: int, group_id: int, prob: float):
icon="sword.png",
)
async def _(user_id: int, group_id: int):
print(user_id, group_id, "使用测试道具")
# print(user_id, group_id, "使用测试道具")
pass
@shop_register.before_handle(name="测试道具A", load_status=False)
async def _(user_id: int, group_id: int):
print(user_id, group_id, "第一个使用前函数（before handle）")
# print(user_id, group_id, "第一个使用前函数（before handle）")
pass
@shop_register.before_handle(name="测试道具A", load_status=False)
async def _(user_id: int, group_id: int):
print(user_id, group_id, "第二个使用前函数（before handle）222")
raise NotMeetUseConditionsException("太笨了！") # 抛出异常，阻断使用，并返回信息
# print(user_id, group_id, "第二个使用前函数（before handle）222")
# raise NotMeetUseConditionsException("太笨了！") # 抛出异常，阻断使用，并返回信息
pass
@shop_register.after_handle(name="测试道具A", load_status=False)
async def _(user_id: int, group_id: int):
print(user_id, group_id, "第一个使用后函数（after handle）")
# print(user_id, group_id, "第一个使用后函数（after handle）")
pass

41
zhenxun/builtin_plugins/web_ui/api/tabs/manage/chat.py
View File

@ -1,20 +1,15 @@
import re
from typing import Literal
import nonebot
from fastapi import APIRouter
from nonebot import on_message
from nonebot.adapters.onebot.v11 import MessageEvent
from nonebot_plugin_alconna import At, Emoji, Hyper, Image, Text, UniMessage, UniMsg
from nonebot_plugin_session import EventSession
from starlette.websockets import WebSocket, WebSocketDisconnect, WebSocketState
from nonebot.adapters.onebot.v11 import MessageEvent
from nonebot_plugin_alconna import At, Text, Hyper, Image, UniMsg
from starlette.websockets import WebSocket, WebSocketState, WebSocketDisconnect
from zhenxun.models.friend_user import FriendUser
from zhenxun.models.group_console import GroupConsole
from zhenxun.models.group_member_info import GroupInfoUser
from zhenxun.utils.depends import UserName
from zhenxun.models.group_member_info import GroupInfoUser
from ....config import AVA_URL, GROUP_AVA_URL
from ....config import AVA_URL
from .model import Message, MessageItem
driver = nonebot.get_driver()
@ -27,7 +22,8 @@ ID_LIST = []
ws_router = APIRouter()
matcher = on_message(block=False, priority=1)
matcher = on_message(block=False, priority=1, rule=lambda: bool(ws_conn))
@driver.on_shutdown
@ -44,7 +40,7 @@ async def _(websocket: WebSocket):
ws_conn = websocket
try:
while websocket.client_state == WebSocketState.CONNECTED:
recv = await websocket.receive()
await websocket.receive()
except WebSocketDisconnect:
ws_conn = None
@ -55,7 +51,7 @@ async def message_handle(
):
messages = []
for m in message:
if isinstance(m, (Text, str)):
if isinstance(m, Text | str):
messages.append(MessageItem(type="text", msg=str(m)))
elif isinstance(m, Image):
if m.url:
@ -70,18 +66,15 @@ async def message_handle(
ID2NAME[group_id] = {}
if m.target in ID2NAME[group_id]:
uname = ID2NAME[group_id][m.target]
else:
if group_user := await GroupInfoUser.get_or_none(
user_id=m.target, group_id=group_id
):
uname = group_user.user_name
if m.target not in ID2NAME[group_id]:
ID2NAME[group_id][m.target] = uname
elif group_user := await GroupInfoUser.get_or_none(
user_id=m.target, group_id=group_id
):
uname = group_user.user_name
if m.target not in ID2NAME[group_id]:
ID2NAME[group_id][m.target] = uname
messages.append(MessageItem(type="at", msg=f"@{uname}"))
# elif isinstance(m, Emoji):
# messages.append(MessageItem(type="text", msg=f"[emoji]"))
elif isinstance(m, Hyper):
messages.append(MessageItem(type="text", msg=f"[分享消息]"))
messages.append(MessageItem(type="text", msg="[分享消息]"))
return messages
@ -91,7 +84,6 @@ async def _(
):
global ws_conn, ID2NAME, ID_LIST
uid = session.id1
gid = session.id3 or session.id2
if ws_conn and ws_conn.client_state == WebSocketState.CONNECTED and uid:
msg_id = event.message_id
if msg_id in ID_LIST:
@ -99,6 +91,7 @@ async def _(
ID_LIST.append(msg_id)
if len(ID_LIST) > 50:
ID_LIST = ID_LIST[40:]
gid = session.id3 or session.id2
messages = await message_handle(message, gid)
data = Message(
object_id=gid or uid,