diff --git a/plugins/genshin/avatar_list.py b/plugins/genshin/avatar_list.py index 12a6f162..ba6c334a 100644 --- a/plugins/genshin/avatar_list.py +++ b/plugins/genshin/avatar_list.py @@ -1,32 +1,28 @@ -"""练度统计""" import asyncio -from typing import List, Optional, Sequence +from typing import List, Optional, Sequence, TYPE_CHECKING -from aiohttp import ClientConnectorError -from arkowrapper import ArkoWrapper -from enkanetwork import Assets as EnkaAssets, EnkaNetworkAPI, VaildateUIDError, HTTPException, EnkaPlayerNotFound from genshin import Client, GenshinException, InvalidCookies from genshin.models import CalculatorCharacterDetails, CalculatorTalent, Character -from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update, User +from telegram import InlineKeyboardButton, InlineKeyboardMarkup from telegram.constants import ChatAction, ParseMode -from telegram.ext import CallbackContext, filters +from telegram.ext import filters from telegram.helpers import create_deep_linked_url -from core.config import config from core.dependence.assets import AssetsService -from core.dependence.redisdb import RedisDB from core.plugin import Plugin, handler from core.services.cookies import CookiesService from core.services.players import PlayersService from core.services.players.services import PlayerInfoService from core.services.template.models import FileType from core.services.template.services import TemplateService -from metadata.genshin import AVATAR_DATA, NAMECARD_DATA +from metadata.genshin import AVATAR_DATA from modules.wiki.base import Model from plugins.tools.genshin import CookiesNotFoundError, GenshinHelper, PlayerNotFoundError, CharacterDetails -from utils.enkanetwork import RedisCache from utils.log import logger -from utils.patch.aiohttp import AioHttpTimeoutException + +if TYPE_CHECKING: + from telegram.ext import ContextTypes + from telegram import Update, User class SkillData(Model): @@ -52,13 +48,14 @@ class AvatarData(Model): class AvatarListPlugin(Plugin): + """练度统计""" + def __init__( self, player_service: PlayersService = None, cookies_service: CookiesService = None, assets_service: AssetsService = None, template_service: TemplateService = None, - redis: RedisDB = None, helper: GenshinHelper = None, character_details: CharacterDetails = None, player_info_service: PlayerInfoService = None, @@ -66,15 +63,12 @@ class AvatarListPlugin(Plugin): self.cookies_service = cookies_service self.assets_service = assets_service self.template_service = template_service - self.enka_client = EnkaNetworkAPI(lang="chs", user_agent=config.enka_network_api_agent) - self.enka_client.set_cache(RedisCache(redis.client, key="plugin:avatar_list:enka_network", ex=60 * 60 * 3)) - self.enka_assets = EnkaAssets(lang="chs") self.helper = helper self.character_details = character_details self.player_service = player_service self.player_info_service = player_info_service - async def get_user_client(self, update: Update, context: CallbackContext) -> Optional[Client]: + async def get_user_client(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE") -> Optional[Client]: message = update.effective_message user = update.effective_user try: @@ -166,43 +160,7 @@ class AvatarListPlugin(Plugin): reverse=True, )[:max_length] - async def get_final_data(self, client: Client, characters: Sequence[Character], update: Update): - try: - response = await self.enka_client.fetch_user(client.uid, info=True) - name_card = (await self.assets_service.namecard(response.player.namecard.id).navbar()).as_uri() - avatar = (await self.assets_service.avatar(response.player.avatar.id).icon()).as_uri() - nickname = response.player.nickname - if response.player.avatar.id in [10000005, 10000007]: - rarity = 5 - else: - rarity = {k: v["rank"] for k, v in AVATAR_DATA.items()}[str(response.player.avatar.id)] - return name_card, avatar, nickname, rarity - except (VaildateUIDError, EnkaPlayerNotFound, HTTPException) as exc: - logger.warning("EnkaNetwork 请求失败: %s", str(exc)) - except (AioHttpTimeoutException, ClientConnectorError) as exc: - logger.warning("EnkaNetwork 请求超时: %s", str(exc)) - except Exception as exc: - logger.error("EnkaNetwork 请求失败: %s", exc_info=exc) - choices = ArkoWrapper(characters).filter(lambda x: x.friendship == 10) # 筛选出好感满了的角色 - if choices.length == 0: # 若没有满好感角色、则以好感等级排序 - choices = ArkoWrapper(characters).sort(lambda x: x.friendship, reverse=True) - name_card_choices = ( # 找到与角色对应的满好感名片ID - ArkoWrapper(choices) - .map(lambda x: next(filter(lambda y: y["name"].split("·")[0] == x.name, NAMECARD_DATA.values()), None)) - .filter(lambda x: x) - .map(lambda x: int(x["id"])) - ) - # noinspection PyTypeChecker - name_card = (await self.assets_service.namecard(name_card_choices[0]).navbar()).as_uri() - avatar = (await self.assets_service.avatar(cid := choices[0].id).icon()).as_uri() - nickname = update.effective_user.full_name - if cid in [10000005, 10000007]: - rarity = 5 - else: - rarity = {k: v["rank"] for k, v in AVATAR_DATA.items()}[str(cid)] - return name_card, avatar, nickname, rarity - - async def get_default_final_data(self, player_id: int, characters: Sequence[Character], user: User): + async def get_final_data(self, player_id: int, user: "User"): player = await self.player_service.get(user.id, player_id) player_info = await self.player_info_service.get(player) nickname = user.full_name @@ -213,22 +171,17 @@ class AvatarListPlugin(Plugin): if player_info.nickname is not None: nickname = player_info.nickname if player_info.name_card is not None: - name_card = (await self.assets_service.namecard(player_info.name_card).navbar()).as_uri() + name_card = (await self.assets_service.namecard(int(player_info.name_card)).navbar()).as_uri() if player_info.hand_image is not None: avatar = (await self.assets_service.avatar(player_info.hand_image).icon()).as_uri() rarity = {k: v["rank"] for k, v in AVATAR_DATA.items()}[str(player_info.hand_image)] - if name_card is not None: # 须弥·正明 - name_card = (await self.assets_service.namecard(210132).navbar()).as_uri() - if avatar is not None: - if traveller := next(filter(lambda x: x.id in [10000005, 10000007], characters), None): - avatar = (await self.assets_service.avatar(traveller.id).icon()).as_uri() - else: - avatar = (await self.assets_service.avatar(10000005).icon()).as_uri() + if name_card is not None: # 默认 + name_card = (await self.assets_service.namecard(210001).navbar()).as_uri() return name_card, avatar, nickname, rarity @handler.command("avatars", filters.Regex(r"^/avatars\s*(?:(\d+)|(all))?$"), block=False) @handler.message(filters.Regex(r"^(全部)?练度统计$"), block=False) - async def avatar_list(self, update: Update, context: CallbackContext): + async def avatar_list(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE"): user = update.effective_user message = update.effective_message @@ -267,11 +220,7 @@ class AvatarListPlugin(Plugin): return raise e - try: - name_card, avatar, nickname, rarity = await self.get_final_data(client, characters, update) - except Exception as exc: # pylint: disable=W0703 - logger.error("卡片信息请求失败 %s", str(exc)) - name_card, avatar, nickname, rarity = await self.get_default_final_data(client.uid, characters, user) + name_card, avatar, nickname, rarity = await self.get_final_data(client.uid, user) render_data = { "uid": client.uid, # 玩家uid