PamGram/plugins/starrail/stats.py

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 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)