import contextlib from typing import Optional, List, TYPE_CHECKING from simnet import Region from simnet.errors import BadRequest as SimnetBadRequest from telegram import Update, Message from telegram.constants import ChatAction from telegram.ext import CallbackContext, filters from core.plugin import Plugin, handler from core.services.cookies.error import TooManyRequestPublicCookies from core.services.players.services import PlayerInfoService from core.services.template.models import RenderResult from core.services.template.services import TemplateService from gram_core.config import config from gram_core.plugin.methods.inline_use_data import IInlineUseData from plugins.tools.genshin import GenshinHelper from plugins.tools.head_icon import HeadIconService from plugins.tools.phone_theme import PhoneThemeService from utils.log import logger from utils.uid import mask_number if TYPE_CHECKING: from simnet import StarRailClient __all__ = ("PlayerStatsPlugins",) class PlayerStatsPlugins(Plugin): """玩家统计查询""" def __init__( self, template: TemplateService, helper: GenshinHelper, head_icon: HeadIconService, phone_theme: PhoneThemeService, player_info_service: PlayerInfoService, ): self.template_service = template self.helper = helper self.head_icon = head_icon self.phone_theme = phone_theme self.player_info_service = player_info_service async def get_uid(self, user_id: int, reply: Optional[Message], player_id: int, offset: int) -> int: """通过消息获取 uid，优先级：args > reply > self""" uid, user_id_ = player_id, user_id 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 @handler.command("stats", player=True, block=False) @handler.message(filters.Regex("^玩家统计查询(.*)"), player=True, block=False) async def command_start(self, update: Update, _: CallbackContext) -> Optional[int]: user_id = await self.get_real_user_id(update) uid, offset = self.get_real_uid_or_offset(update) message = update.effective_message self.log_user(update, logger.info, "查询游戏用户命令请求") try: uid: int = await self.get_uid(user_id, 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, uid) 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"{user_id}.png") async def render(self, client: "StarRailClient", uid: Optional[int] = None) -> RenderResult: if uid is None: uid = client.player_id user_info = await client.get_starrail_user(uid) try: rogue = await client.get_starrail_rogue(uid) except SimnetBadRequest: rogue = None ledger = None if (not client.public) and client.region != Region.OVERSEAS: with contextlib.suppress(SimnetBadRequest): ledger = await client.get_starrail_ledger_month_info(uid) logger.debug(user_info) await self.set_name_card(uid, user_info.phone_background_image_url) data = { "uid": mask_number(uid), "info": user_info.info, "stats": user_info.stats, "stats_labels": [ ("活跃天数", "active_days"), ("获取角色数", "avatar_num"), ("成就达成数", "achievement_num"), ("战利品开启数", "chest_num"), ("逐光捡金", "abyss_process"), ("梦境护照贴纸", "dream_paster_num"), ], "rogue": rogue.basic_info if rogue else None, "rogue_labels": [ ("技能树已激活", "unlocked_skill_points"), ("已解锁奇物", "unlocked_miracle_num"), ("已解锁祝福", "unlocked_buff_num"), ], "ledger": ledger, "ledger_labels": [ ("本月星琼", "current_hcoin"), ("本月通专票", "current_rails_pass"), ("上月星琼", "last_hcoin"), ("上月通专票", "last_rails_pass"), ], "style": "xianzhou", # nosec "avatar": (await self.head_icon.get_head_icon(uid)).as_uri(), "background": (await self.phone_theme.get_phone_theme(uid)).as_uri(), } return await self.template_service.render( "starrail/stats/stats.html", data, {"width": 650, "height": 440}, full_page=True, ) async def set_name_card(self, player_id: int, image_url: str): if not image_url: return try: phone_theme_id = int(image_url.split("/")[-1].replace(".png", "")) except (IndexError, ValueError): return await self.phone_theme.set_to_cache(player_id, phone_theme_id) await self.player_info_service.set_name_card(player_id, phone_theme_id) async def stats_use_by_inline(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE"): callback_query = update.callback_query user = update.effective_user user_id = user.id uid = IInlineUseData.get_uid_from_context(context) self.log_user(update, logger.info, "查询游戏用户命令请求") notice = None try: async with self.helper.genshin_or_public(user_id, uid=uid) as client: if not client.public: await client.get_record_cards() render_result = await self.render(client, client.player_id) except TooManyRequestPublicCookies: notice = "用户查询次数过多 请稍后重试" except AttributeError as exc: logger.error("角色数据有误") logger.exception(exc) notice = f"角色数据有误 估计是{config.notice.bot_name}晕了" except ValueError as exc: logger.warning("获取 uid 发生错误！ 错误信息为 %s", str(exc)) notice = "UID 内部错误" if notice: await callback_query.answer(notice, show_alert=True) return await render_result.edit_inline_media(callback_query) async def get_inline_use_data(self) -> List[Optional[IInlineUseData]]: return [ IInlineUseData( text="玩家统计", hash="stats", callback=self.stats_use_by_inline, player=True, ), ]