mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2025-12-15 06:12:53 +08:00
update ShopRegister
This commit is contained in:
HibiKier
parent
901a90ff13
commit
1f54226673
@ -242,9 +242,10 @@ __Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
|
||||
|
||||
## 更新
|
||||
|
||||
### 2022/5/
|
||||
### 2022/5/3
|
||||
|
||||
* 商品使用函数可以添加参数ShopParam，从中获取user_id等以及自己提供的参数
|
||||
* 商品使用函数可以添加特定参数，例如：user_id, group_id, ShopParam等以及自己提供的参数
|
||||
* 添加商品注册装饰器shop_register
|
||||
|
||||
### 2022/5/1
|
||||
|
||||
|
||||
@ -7,7 +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
|
||||
from utils.decorator.shop import shop_register
|
||||
import nonebot
|
||||
import time
|
||||
|
||||
@ -22,27 +22,25 @@ async def init_default_shop_goods():
|
||||
导入内置的三个商品
|
||||
"""
|
||||
|
||||
# @shop_register(
|
||||
# name="好感度双倍加持卡Ⅰ",
|
||||
# price=30,
|
||||
# des="下次签到双倍好感度概率 + 10%（谁才是真命天子？）（同类商品将覆盖）",
|
||||
# ** {"prob": 0.1}
|
||||
# )
|
||||
@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()
|
||||
|
||||
|
||||
# 创建商店界面
|
||||
@ -64,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"
|
||||
)
|
||||
@ -97,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] + "分"
|
||||
@ -130,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,
|
||||
):
|
||||
"""
|
||||
添加商品
|
||||
@ -159,7 +163,11 @@ async def register_goods(
|
||||
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
|
||||
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
|
||||
)
|
||||
@ -236,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],
|
||||
|
||||
@ -34,7 +34,7 @@ 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(
|
||||
@ -159,7 +159,7 @@ def register_use(goods_name: str, func, **kwargs):
|
||||
raise ValueError("该商品使用函数已被注册！")
|
||||
# 发送使用成功信息
|
||||
kwargs["send_success_msg"] = kwargs.get("send_success_msg", True)
|
||||
kwargs["_max_num_limit"] = kwargs.get("_max_num_limit", 1)
|
||||
kwargs["max_num_limit"] = kwargs.get("max_num_limit", 1)
|
||||
func_manager.register_use(
|
||||
goods_name,
|
||||
**{
|
||||
|
||||
@ -1,60 +1,110 @@
|
||||
#
|
||||
# from functools import wraps
|
||||
# from typing import Union, List, Callable
|
||||
# from nonebot.plugin import require
|
||||
# from nonebot.adapters.onebot.v11 import Bot
|
||||
# import asyncio
|
||||
# import nonebot
|
||||
#
|
||||
# driver = nonebot.get_driver()
|
||||
#
|
||||
# use = require("use")
|
||||
# shop = require("shop_handle")
|
||||
#
|
||||
# flag = False
|
||||
#
|
||||
# name_list = []
|
||||
#
|
||||
# func_list = []
|
||||
#
|
||||
#
|
||||
# def shop_register(
|
||||
# name: Union[str, List[str]],
|
||||
# price: Union[int, List[int]],
|
||||
# des: Union[str, List[str]],
|
||||
# discount: Union[float, List[float]] = 1,
|
||||
# limit_time: Union[int, List[int]] = 0,
|
||||
# status: bool = True,
|
||||
# **kwargs_
|
||||
# ):
|
||||
# print("---------")
|
||||
# print("name：", name)
|
||||
# print("price：", price)
|
||||
# print("des：", des)
|
||||
# print("discount：", discount)
|
||||
# print("limit_time：", limit_time)
|
||||
# print("status：", status)
|
||||
# print("kwargs:", kwargs_)
|
||||
# asyncio.run(shop.register_goods(
|
||||
# name, 30, price, discount, limit_time
|
||||
# ))
|
||||
#
|
||||
# def _register_use(goods_func: Callable):
|
||||
# def _wrapper(**kwargs):
|
||||
# # print(*args)
|
||||
# print(**kwargs)
|
||||
# print(1111111111111111)
|
||||
# use.register_use(name, goods_func, **kwargs)
|
||||
# # func_list.append({"name": name, "func": goods_func, "args": args, "kwargs": kwargs})
|
||||
# return _wrapper
|
||||
#
|
||||
# return _register_use
|
||||
#
|
||||
#
|
||||
# @driver.on_bot_connect
|
||||
# async def do_something(bot: Bot):
|
||||
# for func in func_list:
|
||||
# if asyncio.iscoroutinefunction(func):
|
||||
# await func()
|
||||
# else:
|
||||
# func()
|
||||
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()
|
||||
|
||||
@ -11,4 +11,4 @@ class ShopParam(BaseModel):
|
||||
event: MessageEvent
|
||||
num: int # 道具单次使用数量
|
||||
send_success_msg: bool = True # 是否发送使用成功信息
|
||||
_max_num_limit: int = 1 # 单次使用最大次数
|
||||
max_num_limit: int = 1 # 单次使用最大次数
|
||||
|
||||
Loading…
Reference in New Issue
Block a user