from typing import Optional, List from genshin import Client, GenshinException from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup, Message from telegram.constants import ChatAction from telegram.ext import CallbackContext, filters from telegram.helpers import create_deep_linked_url 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, PlayerNotFoundError, CookiesNotFoundError from utils.log import logger __all__ = ("PlayerStatsPlugins",) class PlayerStatsPlugins(Plugin): """玩家统计查询""" def __init__( self, template: TemplateService, helper: GenshinHelper, ): self.template_service = template self.helper = helper async def get_uid(self, user_id: int, args: List[str], reply: Optional[Message]) -> int: """通过消息获取 uid,优先级:args > reply > self""" uid, user_id_ = None, user_id if args: for i in args: if i is not None: if i.isdigit() and len(i) == 9: uid = int(i) 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_) 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) if player_info is not None: uid = player_info.player_id return uid @handler.command("stats", block=False) @handler.message(filters.Regex("^玩家统计查询(.*)"), block=False) async def command_start(self, update: Update, context: CallbackContext) -> Optional[int]: user = update.effective_user message = update.effective_message logger.info("用户 %s[%s] 查询游戏用户命令请求", user.full_name, user.id) try: uid: int = await self.get_uid(user.id, context.args, message.reply_to_message) try: client = await self.helper.get_genshin_client(user.id) if client.uid != uid: raise CookiesNotFoundError except CookiesNotFoundError: client, _ = await self.helper.get_public_genshin_client(user.id) render_result = await self.render(client, uid) except PlayerNotFoundError: buttons = [[InlineKeyboardButton("点我绑定账号", url=create_deep_linked_url(context.bot.username, "set_cookie"))]] if filters.ChatType.GROUPS.filter(message): reply_message = await message.reply_text( "未查询到您所绑定的账号信息,请先私聊彦卿绑定账号", reply_markup=InlineKeyboardMarkup(buttons) ) self.add_delete_message_job(reply_message, delay=30) self.add_delete_message_job(message, delay=30) else: await message.reply_text("未查询到您所绑定的账号信息,请先绑定账号", reply_markup=InlineKeyboardMarkup(buttons)) return except GenshinException as exc: if exc.retcode == 1034: await message.reply_text("出错了呜呜呜 ~ 请稍后重试") return raise exc except TooManyRequestPublicCookies: await message.reply_text("用户查询次数过多 请稍后重试") return 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"{client.uid}.png", allow_sending_without_reply=True) async def render(self, client: Client, uid: Optional[int] = None) -> RenderResult: if uid is None: uid = client.uid user_info = await client.get_starrail_user(uid) try: rogue = await client.get_starrail_rogue(uid) except GenshinException: rogue = None logger.debug(user_info) # 因为需要替换线上图片地址为本地地址,先克隆数据,避免修改原数据 user_info = user_info.copy(deep=True) data = { "uid": uid, "info": user_info.info, "stats": user_info.stats, "stats_labels": [ ("活跃天数", "active_days"), ("成就达成数", "achievement_num"), ("获取角色数", "avatar_num"), ("忘却之庭", "abyss_process"), ("战利品开启数", "chest_num"), ], "rogue": rogue.basic_info if rogue else None, "rogue_labels": [ ("技能树已激活", "unlocked_skill_points"), ("已解锁奇物", "unlocked_miracle_num"), ("已解锁祝福", "unlocked_buff_num"), ], "style": "xianzhou", # nosec } return await self.template_service.render( "starrail/stats/stats.html", data, {"width": 650, "height": 440}, full_page=True, )