PaiGram/plugins/genshin/wish.py

325 lines
15 KiB
Python
Raw Normal View History

2022-10-15 13:06:42 +00:00
import asyncio
import re
2022-10-14 04:05:40 +00:00
from datetime import datetime
from typing import Any, List, Optional, Tuple, Union
2022-07-31 08:15:09 +00:00
from bs4 import BeautifulSoup
2022-07-31 08:15:09 +00:00
from telegram import Update
from telegram.constants import ChatAction
from telegram.ext import CallbackContext, CommandHandler, MessageHandler, filters
2022-07-31 08:15:09 +00:00
from core.dependence.assets import AssetsService
from core.dependence.redisdb import RedisDB
from core.plugin import Plugin, handler
from core.services.template.services import TemplateService
from metadata.genshin import AVATAR_DATA, WEAPON_DATA, avatar_to_game_id, weapon_to_game_id
2022-10-15 13:06:42 +00:00
from metadata.shortname import weaponToName
2022-12-10 12:37:43 +00:00
from modules.apihelper.client.components.gacha import Gacha as GachaClient
from modules.apihelper.models.genshin.gacha import GachaInfo
from modules.wish.banner import GenshinBannerType, GachaBanner
from modules.wish.player.info import PlayerGachaInfo
from modules.wish.system import BannerSystem
from utils.log import logger
2022-07-31 08:15:09 +00:00
try:
import ujson as jsonlib
except ImportError:
import json as jsonlib
2022-07-31 08:15:09 +00:00
class GachaNotFound(Exception):
"""卡池未找到"""
2022-10-15 13:06:42 +00:00
def __init__(self, gacha_name: str):
self.gacha_name = gacha_name
super().__init__(f"{gacha_name} gacha not found")
class GachaDataFound(Exception):
"""卡池数据未找到"""
def __init__(self, item_id: int):
self.item_id = item_id
super().__init__(f"item_id[{item_id}] data not found")
2022-10-15 13:06:42 +00:00
class GachaRedis:
def __init__(self, redis: RedisDB):
self.client = redis.client
self.qname = "plugin:gacha:"
2022-07-31 08:15:09 +00:00
2022-10-15 13:06:42 +00:00
async def get(self, user_id: int) -> PlayerGachaInfo:
data = await self.client.get(f"{self.qname}{user_id}")
if data is None:
return PlayerGachaInfo()
return PlayerGachaInfo(**jsonlib.loads(data))
2022-07-31 08:15:09 +00:00
2022-10-15 13:06:42 +00:00
async def set(self, user_id: int, player_gacha_info: PlayerGachaInfo):
value = player_gacha_info.json()
await self.client.set(f"{self.qname}{user_id}", value)
class WishSimulatorHandle:
2022-12-10 12:37:43 +00:00
def __init__(self):
self.hyperion = GachaClient()
2022-10-15 13:06:42 +00:00
async def de_banner(self, gacha_id: str, gacha_type: int) -> Optional[GachaBanner]:
gacha_info = await self.hyperion.get_gacha_info(gacha_id)
banner = GachaBanner()
banner.banner_id = gacha_id
2022-10-15 13:06:42 +00:00
banner.title, banner.html_title = self.de_title(gacha_info["title"])
r5_up_items = gacha_info.get("r5_up_items")
if r5_up_items is not None:
for r5_up_item in r5_up_items:
if r5_up_item["item_type"] == "角色":
banner.rate_up_items5.append(avatar_to_game_id(r5_up_item["item_name"]))
elif r5_up_item["item_type"] == "武器":
banner.rate_up_items5.append(weapon_to_game_id(r5_up_item["item_name"]))
r5_prob_list = gacha_info.get("r5_prob_list")
if r5_prob_list is not None:
for r5_prob in gacha_info.get("r5_prob_list", []):
if r5_prob["item_type"] == "角色":
banner.fallback_items5_pool1.append(avatar_to_game_id(r5_prob["item_name"]))
elif r5_prob["item_type"] == "武器":
banner.fallback_items5_pool1.append(weapon_to_game_id(r5_prob["item_name"]))
r4_up_items = gacha_info.get("r4_up_items")
if r4_up_items is not None:
for r4_up_item in r4_up_items:
if r4_up_item["item_type"] == "角色":
banner.rate_up_items4.append(avatar_to_game_id(r4_up_item["item_name"]))
elif r4_up_item["item_type"] == "武器":
banner.rate_up_items4.append(weapon_to_game_id(r4_up_item["item_name"]))
r4_prob_list = gacha_info.get("r4_prob_list")
if r4_prob_list is not None:
for r4_prob in r4_prob_list:
if r4_prob["item_type"] == "角色":
banner.fallback_items4_pool1.append(avatar_to_game_id(r4_prob["item_name"]))
elif r4_prob["item_type"] == "武器":
banner.fallback_items4_pool1.append(weapon_to_game_id(r4_prob["item_name"]))
2022-10-22 13:54:04 +00:00
if gacha_type in {301, 400}:
2022-10-15 13:06:42 +00:00
banner.wish_max_progress = 1
banner.banner_type = GenshinBannerType.EVENT
2022-10-15 13:06:42 +00:00
banner.weight4 = ((1, 510), (8, 510), (10, 10000))
banner.weight5 = ((1, 60), (73, 60), (90, 10000))
elif gacha_type == 302:
banner.wish_max_progress = 2
banner.banner_type = GenshinBannerType.WEAPON
2022-10-15 13:06:42 +00:00
banner.weight4 = ((1, 600), (7, 600), (10, 10000))
banner.weight5 = ((1, 70), (62, 70), (90, 10000))
else:
banner.banner_type = GenshinBannerType.STANDARD
2022-10-15 13:06:42 +00:00
return banner
2022-12-10 12:37:43 +00:00
async def gacha_base_info(self, gacha_name: str = "角色活动", default: bool = False) -> GachaInfo:
2022-10-15 13:06:42 +00:00
gacha_list_info = await self.hyperion.get_gacha_list_info()
2022-10-14 04:05:40 +00:00
now = datetime.now()
for gacha in gacha_list_info:
if gacha.gacha_name == gacha_name and gacha.begin_time <= now <= gacha.end_time:
2022-10-15 13:06:42 +00:00
return gacha
else: # pylint: disable=W0120
2022-10-14 04:05:40 +00:00
if default and len(gacha_list_info) > 0:
2022-10-15 13:06:42 +00:00
return gacha_list_info[0]
raise GachaNotFound(gacha_name)
2022-10-15 13:06:42 +00:00
@staticmethod
def de_title(title: str) -> Union[Tuple[str, None], Tuple[str, Any]]:
2022-10-15 13:06:42 +00:00
title_html = BeautifulSoup(title, "lxml")
re_color = re.search(r"<color=#(.*?)>", title, flags=0)
if re_color is None:
return title_html.text, None
2022-10-22 13:54:04 +00:00
color = re_color[1]
2022-10-15 13:06:42 +00:00
title_html.color.name = "span"
title_html.span["style"] = f"color:#{color};"
return title_html.text, title_html.p
class WishSimulatorPlugin(Plugin):
2022-10-15 13:06:42 +00:00
"""抽卡模拟器(非首模拟器/减寿模拟器)"""
def __init__(self, assets: AssetsService, template_service: TemplateService, redis: RedisDB):
2022-10-15 13:06:42 +00:00
self.gacha_db = GachaRedis(redis)
self.handle = WishSimulatorHandle()
2022-10-15 13:06:42 +00:00
self.banner_system = BannerSystem()
self.template_service = template_service
self.banner_cache = {}
self._look = asyncio.Lock()
self.assets_service = assets
2022-10-15 13:06:42 +00:00
2022-12-10 12:37:43 +00:00
async def get_banner(self, gacha_base_info: GachaInfo):
2022-10-15 13:06:42 +00:00
async with self._look:
banner = self.banner_cache.get(gacha_base_info.gacha_id)
if banner is None:
banner = await self.handle.de_banner(gacha_base_info.gacha_id, gacha_base_info.gacha_type)
self.banner_cache.setdefault(gacha_base_info.gacha_id, banner)
return banner
2022-07-31 08:15:09 +00:00
2022-11-23 01:11:29 +00:00
async def de_item_list(self, item_list: List[int]) -> List[dict]:
gacha_item: List[dict] = []
for item_id in item_list:
if item_id is None:
continue
if 10000 <= item_id <= 100000:
data = WEAPON_DATA.get(str(item_id))
avatar = self.assets_service.weapon(item_id)
gacha = await avatar.gacha()
if gacha is None:
raise GachaDataFound(item_id)
data.setdefault("url", gacha.as_uri())
gacha_item.append(data)
elif 10000000 <= item_id <= 19999999:
data = AVATAR_DATA.get(str(item_id))
avatar = self.assets_service.avatar(item_id)
gacha = await avatar.gacha_card()
if gacha is None:
raise GachaDataFound(item_id)
data.setdefault("url", gacha.as_uri())
gacha_item.append(data)
return gacha_item
async def shutdown(self) -> None:
pass
# todo 目前清理消息无法执行 因为先停止Job导致无法获取全部信息
# logger.info("正在清理消息")
# job_queue = self.application.telegram.job_queue
# jobs = job_queue.jobs()
# for job in jobs:
# if "wish_simulator" in job.name and not job.removed:
# logger.info("当前Job name %s", job.name)
# try:
# await job.run(job_queue.application)
# except CancelledError:
# continue
# except Exception as exc:
# logger.warning("执行失败 %", str(exc))
# else:
# logger.info("Jobs为空")
# logger.success("清理卡池消息成功")
2022-12-01 02:33:09 +00:00
@handler(CommandHandler, command="wish", block=False)
@handler(MessageHandler, filters=filters.Regex("^抽卡模拟器(.*)"), block=False)
2022-07-31 08:15:09 +00:00
async def command_start(self, update: Update, context: CallbackContext) -> None:
message = update.effective_message
2022-07-31 08:15:09 +00:00
user = update.effective_user
args = self.get_args(context)
2022-07-31 08:15:09 +00:00
gacha_name = "角色活动"
if len(args) >= 1:
gacha_name = args[0]
if gacha_name not in ("角色活动-2", "武器活动", "常驻", "角色活动"):
for key, value in {"2": "角色活动-2", "武器": "武器活动", "普通": "常驻"}.items():
if key == gacha_name:
gacha_name = value
break
try:
2022-10-15 13:06:42 +00:00
gacha_base_info = await self.handle.gacha_base_info(gacha_name)
except GachaNotFound as exc:
2022-10-22 13:54:04 +00:00
await message.reply_text(f"没有找到名为 {exc.gacha_name} 的卡池,可能是卡池不存在或者卡池已经结束,请检查后重试。如果你想抽取默认卡池,请不要输入参数。")
2022-07-31 08:15:09 +00:00
return
else:
try:
gacha_base_info = await self.handle.gacha_base_info(default=True)
except GachaNotFound:
await message.reply_text("当前卡池正在替换中,请稍后重试。")
return
2022-12-10 12:37:43 +00:00
logger.info("用户 %s[%s] 抽卡模拟器命令请求 || 参数 %s", user.full_name, user.id, gacha_name)
2022-07-31 08:15:09 +00:00
# 用户数据储存和处理
await message.reply_chat_action(ChatAction.TYPING)
2022-10-15 13:06:42 +00:00
banner = await self.get_banner(gacha_base_info)
player_gacha_info = await self.gacha_db.get(user.id)
# 检查 wish_item_id
2022-10-22 13:54:04 +00:00
if (
banner.banner_type == GenshinBannerType.WEAPON
2022-10-22 13:54:04 +00:00
and player_gacha_info.event_weapon_banner.wish_item_id not in banner.rate_up_items5
):
player_gacha_info.event_weapon_banner.wish_item_id = 0
2022-10-15 13:06:42 +00:00
# 执行抽卡
item_list = self.banner_system.do_pulls(player_gacha_info, banner, 10)
try:
data = await self.de_item_list(item_list)
except GachaDataFound as exc:
logger.warning("角色 item_id[%s] 抽卡立绘未找到", exc.item_id)
reply_message = await message.reply_text("出错了呜呜呜 ~ 卡池部分数据未找到!")
if filters.ChatType.GROUPS.filter(message):
self.add_delete_message_job(reply_message, name="wish_simulator")
self.add_delete_message_job(message, name="wish_simulator")
return
2022-10-15 13:06:42 +00:00
player_gacha_banner_info = player_gacha_info.get_banner_info(banner)
template_data = {
2022-07-31 08:15:09 +00:00
"name": f"{user.full_name}",
"info": gacha_name,
"banner_name": banner.html_title if banner.html_title else banner.title,
2022-10-15 13:06:42 +00:00
"banner_type": banner.banner_type.name,
"player_gacha_banner_info": player_gacha_banner_info,
2022-07-31 08:15:09 +00:00
"items": [],
2022-10-15 13:06:42 +00:00
"wish_name": "",
2022-07-31 08:15:09 +00:00
}
2022-10-15 13:06:42 +00:00
if player_gacha_banner_info.wish_item_id != 0:
weapon = WEAPON_DATA.get(str(player_gacha_banner_info.wish_item_id))
if weapon is not None:
template_data["wish_name"] = weapon["name"]
await self.gacha_db.set(user.id, player_gacha_info)
2022-07-31 08:15:09 +00:00
def take_rang(elem: dict):
return elem["rank"]
2022-10-15 13:06:42 +00:00
data.sort(key=take_rang, reverse=True)
template_data["items"] = data
2022-07-31 08:15:09 +00:00
await message.reply_chat_action(ChatAction.UPLOAD_PHOTO)
png_data = await self.template_service.render(
"genshin/wish/wish.jinja2", template_data, {"width": 1157, "height": 603}, False
)
2022-07-31 08:15:09 +00:00
reply_message = await message.reply_photo(png_data.photo)
2022-07-31 08:15:09 +00:00
if filters.ChatType.GROUPS.filter(message):
self.add_delete_message_job(reply_message, name="wish_simulator")
self.add_delete_message_job(message, name="wish_simulator")
2022-10-15 13:06:42 +00:00
@handler(CommandHandler, command="set_wish", block=False)
@handler(MessageHandler, filters=filters.Regex("^非首模拟器定轨(.*)"), block=False)
async def set_wish(self, update: Update, context: CallbackContext) -> None:
message = update.effective_message
user = update.effective_user
args = self.get_args(context)
2022-10-22 13:54:04 +00:00
try:
gacha_base_info = await self.handle.gacha_base_info("武器活动")
except GachaNotFound:
reply_message = await message.reply_text("当前还没有武器正在 UP可能是卡池不存在或者卡池已经结束。")
if filters.ChatType.GROUPS.filter(reply_message):
self.add_delete_message_job(message, delay=30)
self.add_delete_message_job(reply_message, delay=30)
2022-10-22 13:54:04 +00:00
return
2022-10-15 13:06:42 +00:00
banner = await self.get_banner(gacha_base_info)
2022-10-22 13:54:04 +00:00
up_weapons = {}
for rate_up_items5 in banner.rate_up_items5:
weapon = WEAPON_DATA.get(str(rate_up_items5))
if weapon is None:
continue
up_weapons[weapon["name"]] = rate_up_items5
up_weapons_text = "当前 UP 武器有:" + "".join(up_weapons.keys())
2022-10-15 13:06:42 +00:00
if len(args) >= 1:
weapon_name = args[0]
else:
2022-10-22 13:54:04 +00:00
reply_message = await message.reply_text(f"输入的参数不正确,请输入需要定轨的武器名称。\n{up_weapons_text}")
2022-10-15 13:06:42 +00:00
if filters.ChatType.GROUPS.filter(reply_message):
self.add_delete_message_job(message, delay=30)
self.add_delete_message_job(reply_message, delay=30)
2022-10-15 13:06:42 +00:00
return
weapon_name = weaponToName(weapon_name)
player_gacha_info = await self.gacha_db.get(user.id)
2022-10-22 13:54:04 +00:00
if weapon_name in up_weapons:
player_gacha_info.event_weapon_banner.wish_item_id = up_weapons[weapon_name]
player_gacha_info.event_weapon_banner.failed_chosen_item_pulls = 0
2022-10-15 13:06:42 +00:00
else:
2022-10-22 13:54:04 +00:00
reply_message = await message.reply_text(
f"输入的参数不正确,可能是没有名为 {weapon_name} 的武器或该武器不存在当前 UP 卡池中\n{up_weapons_text}"
)
2022-10-15 13:06:42 +00:00
if filters.ChatType.GROUPS.filter(reply_message):
self.add_delete_message_job(message, delay=30)
self.add_delete_message_job(reply_message, delay=30)
2022-10-15 13:06:42 +00:00
return
await self.gacha_db.set(user.id, player_gacha_info)
reply_message = await message.reply_text(f"抽卡模拟器定轨 {weapon_name} 武器成功")
if filters.ChatType.GROUPS.filter(reply_message):
self.add_delete_message_job(message, delay=30)
self.add_delete_message_job(reply_message, delay=30)