from typing import Optional, List, Dict, Tuple, TYPE_CHECKING from simnet.errors import BadRequest as SimnetBadRequest from simnet.models.starrail.character import RogueCharacter from simnet.models.starrail.chronicle.rogue import StarRailRogue, StarRailRogueLocust from telegram import Update, Message from telegram.constants import ChatAction from telegram.ext import CallbackContext, filters from core.dependence.assets import AssetsService from core.plugin import Plugin, handler from core.services.cookies.error import TooManyRequestPublicCookies from core.services.template.models import RenderResult from core.services.template.services import TemplateService from plugins.tools.genshin import GenshinHelper from utils.log import logger from utils.uid import mask_number if TYPE_CHECKING: from simnet import StarRailClient __all__ = ("PlayerRoguePlugins",) class NotSupport(Exception): """不支持的服务器""" class NotHaveData(Exception): """没有数据""" class PlayerRoguePlugins(Plugin): """玩家模拟宇宙信息查询""" LUO_MA = ["", "Ⅰ", "Ⅱ", "Ⅲ", "Ⅳ", "Ⅴ", "Ⅵ"] BUFF_EN = { "「丰饶」": "Abundance", "「毁灭」": "Destruction", "「智识」": "Erudition", "「同协」": "Harmony", "「巡猎」": "Hunt", "「欢愉」": "Joy", "「记忆」": "Memory", "「虚无」": "Nihility", "「存护」": "Preservation", "「繁育」": "Propagation", } def __init__( self, template: TemplateService, assets: AssetsService, helper: GenshinHelper, ): self.template_service = template self.assets = assets self.helper = helper async def get_uid( self, user_id: int, args: List[str], reply: Optional[Message], player_id: int, offset: int ) -> Tuple[int, bool]: """通过消息获取 uid，优先级：args > reply > self""" uid, user_id_, pre = player_id, user_id, False if args: for i in args: if i is not None and "上" in i: pre = True if reply: try: user_id_ = reply.from_user.id except AttributeError: pass if not uid: player_info = await self.helper.players_service.get_player(user_id_, offset=offset) if player_info is not None: uid = player_info.player_id if (not uid) and (user_id_ != user_id): player_info = await self.helper.players_service.get_player(user_id, offset=offset) if player_info is not None: uid = player_info.player_id return uid, pre @handler.command("rogue", block=False) @handler.message(filters.Regex("^模拟宇宙信息查询(.*)"), block=False) async def command_start(self, update: Update, context: CallbackContext) -> Optional[int]: user_id = await self.get_real_user_id(update) message = update.effective_message uid, offset = self.get_real_uid_or_offset(update) self.log_user(update, logger.info, "查询模拟宇宙信息命令请求") try: uid, pre = await self.get_uid(user_id, context.args, message.reply_to_message, uid, offset) async with self.helper.genshin_or_public(user_id, uid=uid) as client: render_result = await self.render(client, pre, uid) except AttributeError as exc: logger.error("模拟宇宙数据有误") logger.exception(exc) await message.reply_text("模拟宇宙数据有误 估计是彦卿晕了") return except NotHaveData: reply_message = await message.reply_text("没有查找到模拟宇宙数据") if filters.ChatType.GROUPS.filter(reply_message): self.add_delete_message_job(message) self.add_delete_message_job(reply_message) return await message.reply_chat_action(ChatAction.UPLOAD_PHOTO) await render_result.reply_photo(message, filename=f"{user_id}.png") async def get_rander_data(self, uid: int, data: StarRailRogue, pre: bool) -> Dict: record_raw = data.last_record if pre else data.current_record if not record_raw.has_data: raise NotHaveData record = record_raw.records[0] avatars = record.final_lineup new_avatars = [None, None, None, None] for idx, avatar in enumerate(avatars): old_avatar = avatar.dict() old_avatar["icon"] = self.assets.avatar.square(avatar.id).as_uri() new_avatars[idx] = RogueCharacter(**old_avatar) return { "uid": mask_number(uid), "basic": data.basic_info, "name": f"{record.name} {self.LUO_MA[record.difficulty]}", "finish_cnt": record_raw.basic.finish_cnt, "time": record.finish_time.datetime.strftime("%Y-%m-%d %H:%M"), "score": record.score, "avatars": new_avatars, "buffs": record.buffs, "buff_en_map": self.BUFF_EN, "miracles": record.miracles, } async def render(self, client: "StarRailClient", pre: bool, uid: Optional[int] = None) -> RenderResult: if uid is None: uid = client.player_id rogue = await client.get_starrail_rogue(uid) data = await self.get_rander_data(uid, rogue, pre) return await self.template_service.render( "starrail/rogue/rogue.html", data, {"width": 520, "height": 1000}, full_page=True, query_selector="#new-container", ) @handler.command("rogue_tourn", block=False) @handler.message(filters.Regex("^差分宇宙信息查询(.*)"), block=False) async def rogue_tourn_command_start(self, update: Update, _: CallbackContext) -> Optional[int]: user_id = await self.get_real_user_id(update) message = update.effective_message uid, offset = self.get_real_uid_or_offset(update) self.log_user(update, logger.info, "查询差分宇宙信息命令请求") try: async with self.helper.genshin_or_public(user_id, uid=uid, offset=offset) as client: render_result = await self.render_rogue_tourn(client) except AttributeError as exc: logger.error("差分宇宙数据有误") logger.exception(exc) await message.reply_text("差分宇宙数据有误 估计是彦卿晕了") return await message.reply_chat_action(ChatAction.UPLOAD_PHOTO) await render_result.reply_photo(message, filename=f"{user_id}.png") async def render_rogue_tourn(self, client: "StarRailClient") -> RenderResult: rogue = await client.get_starrail_rogue_tourn() data = { "uid": mask_number(client.player_id), "basic": rogue.basic, } return await self.template_service.render( "starrail/rogue/rogue_tourn.html", data, {"width": 1000, "height": 700}, full_page=True, query_selector=".simulate", ) @handler.command("rogue_locust", block=False) @handler.message(filters.Regex("^寰宇蝗灾信息查询(.*)"), block=False) async def rogue_locust_command_start(self, update: Update, _: CallbackContext) -> Optional[int]: user = update.effective_user message = update.effective_message logger.info("用户 %s[%s] 查询寰宇蝗灾信息命令请求", user.full_name, user.id) try: async with self.helper.genshin(user.id) as client: data = await client.get_starrail_rogue_locust() render_result = await self.rogue_locust_render(data, client.player_id) except SimnetBadRequest as exc: raise exc except AttributeError as exc: logger.error("寰宇蝗灾数据有误") logger.exception(exc) await message.reply_text("寰宇蝗灾数据有误 估计是彦卿晕了") return await message.reply_chat_action(ChatAction.UPLOAD_PHOTO) await render_result.reply_photo(message, filename=f"{user.id}.png") async def rogue_locust_render(self, source: StarRailRogueLocust, uid: int) -> RenderResult: try: record = max(source.detail.records, key=lambda x: x.finish_time.datetime) name = f"{record.name} {self.LUO_MA[record.difficulty]}" new_avatars = [None, None, None, None] for idx, avatar in enumerate(record.final_lineup): old_avatar = avatar.dict() old_avatar["icon"] = self.assets.avatar.square(avatar.id).as_uri() new_avatars[idx] = RogueCharacter(**old_avatar) except ValueError: record, name, new_avatars = None, None, None data = { "uid": mask_number(uid), "cnt": source.basic.cnt, "finish_cnt": len(source.detail.records), "record": record, "name": name, "avatars": new_avatars, "buff_en_map": self.BUFF_EN, } return await self.template_service.render( "starrail/rogue/rogue_locust.html", data, {"width": 520, "height": 1000}, full_page=True, query_selector="#new-locust-container", )