🎨 Update get_final_data

This commit is contained in:
洛水居室 2023-04-28 09:29:16 +08:00
parent fbe49453c7
commit 85178d34ae
No known key found for this signature in database
GPG Key ID: C9DE87DA724B88FC

View File

@ -1,32 +1,28 @@
"""练度统计"""
import asyncio 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 import Client, GenshinException, InvalidCookies
from genshin.models import CalculatorCharacterDetails, CalculatorTalent, Character 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.constants import ChatAction, ParseMode
from telegram.ext import CallbackContext, filters from telegram.ext import filters
from telegram.helpers import create_deep_linked_url from telegram.helpers import create_deep_linked_url
from core.config import config
from core.dependence.assets import AssetsService from core.dependence.assets import AssetsService
from core.dependence.redisdb import RedisDB
from core.plugin import Plugin, handler from core.plugin import Plugin, handler
from core.services.cookies import CookiesService from core.services.cookies import CookiesService
from core.services.players import PlayersService from core.services.players import PlayersService
from core.services.players.services import PlayerInfoService from core.services.players.services import PlayerInfoService
from core.services.template.models import FileType from core.services.template.models import FileType
from core.services.template.services import TemplateService 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 modules.wiki.base import Model
from plugins.tools.genshin import CookiesNotFoundError, GenshinHelper, PlayerNotFoundError, CharacterDetails from plugins.tools.genshin import CookiesNotFoundError, GenshinHelper, PlayerNotFoundError, CharacterDetails
from utils.enkanetwork import RedisCache
from utils.log import logger 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): class SkillData(Model):
@ -52,13 +48,14 @@ class AvatarData(Model):
class AvatarListPlugin(Plugin): class AvatarListPlugin(Plugin):
"""练度统计"""
def __init__( def __init__(
self, self,
player_service: PlayersService = None, player_service: PlayersService = None,
cookies_service: CookiesService = None, cookies_service: CookiesService = None,
assets_service: AssetsService = None, assets_service: AssetsService = None,
template_service: TemplateService = None, template_service: TemplateService = None,
redis: RedisDB = None,
helper: GenshinHelper = None, helper: GenshinHelper = None,
character_details: CharacterDetails = None, character_details: CharacterDetails = None,
player_info_service: PlayerInfoService = None, player_info_service: PlayerInfoService = None,
@ -66,15 +63,12 @@ class AvatarListPlugin(Plugin):
self.cookies_service = cookies_service self.cookies_service = cookies_service
self.assets_service = assets_service self.assets_service = assets_service
self.template_service = template_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.helper = helper
self.character_details = character_details self.character_details = character_details
self.player_service = player_service self.player_service = player_service
self.player_info_service = player_info_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 message = update.effective_message
user = update.effective_user user = update.effective_user
try: try:
@ -166,43 +160,7 @@ class AvatarListPlugin(Plugin):
reverse=True, reverse=True,
)[:max_length] )[:max_length]
async def get_final_data(self, client: Client, characters: Sequence[Character], update: Update): async def get_final_data(self, player_id: int, user: "User"):
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):
player = await self.player_service.get(user.id, player_id) player = await self.player_service.get(user.id, player_id)
player_info = await self.player_info_service.get(player) player_info = await self.player_info_service.get(player)
nickname = user.full_name nickname = user.full_name
@ -213,22 +171,17 @@ class AvatarListPlugin(Plugin):
if player_info.nickname is not None: if player_info.nickname is not None:
nickname = player_info.nickname nickname = player_info.nickname
if player_info.name_card is not None: 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: if player_info.hand_image is not None:
avatar = (await self.assets_service.avatar(player_info.hand_image).icon()).as_uri() 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)] rarity = {k: v["rank"] for k, v in AVATAR_DATA.items()}[str(player_info.hand_image)]
if name_card is not None: # 须弥·正明 if name_card is not None: # 默认
name_card = (await self.assets_service.namecard(210132).navbar()).as_uri() name_card = (await self.assets_service.namecard(210001).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()
return name_card, avatar, nickname, rarity return name_card, avatar, nickname, rarity
@handler.command("avatars", filters.Regex(r"^/avatars\s*(?:(\d+)|(all))?$"), block=False) @handler.command("avatars", filters.Regex(r"^/avatars\s*(?:(\d+)|(all))?$"), block=False)
@handler.message(filters.Regex(r"^(全部)?练度统计$"), 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 user = update.effective_user
message = update.effective_message message = update.effective_message
@ -267,11 +220,7 @@ class AvatarListPlugin(Plugin):
return return
raise e raise e
try: name_card, avatar, nickname, rarity = await self.get_final_data(client.uid, user)
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)
render_data = { render_data = {
"uid": client.uid, # 玩家uid "uid": client.uid, # 玩家uid