diff --git a/modules/apihelper/client/components/akasha.py b/modules/apihelper/client/components/akasha.py new file mode 100644 index 00000000..65684709 --- /dev/null +++ b/modules/apihelper/client/components/akasha.py @@ -0,0 +1,138 @@ +from types import TracebackType +from typing import Optional, Type, List +from urllib.parse import unquote + +import httpx + +from modules.apihelper.models.genshin.akasha import ( + AkashaRank, + AkashaLeaderboardCategory, + AkashaLeaderboard, + AkashaSubStat, + AkashaArtifact, +) + +BASE_URL = "https://akasha.cv/api" +MAIN_API = BASE_URL + "/filters/accounts/" +RANK_API = BASE_URL + "/getCalculationsForUser/" +DATA_API = BASE_URL + "/user/" +REFRESH_API = BASE_URL + "/user/refresh/" +LEADERBOARD_API = BASE_URL + "/leaderboards" +LEADERBOARD_CATEGORY_API = BASE_URL + "/v2/leaderboards/categories" +ARTIFACTS_API = BASE_URL + "/artifacts" + + +class Akasha: + SUB_STAT_MAP = { + AkashaSubStat.CRR: "critValue", + AkashaSubStat.ATK: "substats.ATK%", + AkashaSubStat.HP: "substats.HP%", + AkashaSubStat.DEF: "substats.DEF%", + AkashaSubStat.ATKF: "substats.Flat ATK", + AkashaSubStat.HPF: "substats.Flat HP", + AkashaSubStat.DEFF: "substats.Flat DEF", + AkashaSubStat.EM: "substats.Elemental Mastery", + AkashaSubStat.ER: "substats.Energy Recharge", + AkashaSubStat.CR: "substats.Crit RATE", + AkashaSubStat.CD: "substats.Crit DMG", + } + SUB_STAT_NAME_MAP = { + "Flat ATK": "攻击力", + "Flat HP": "血量", + "Flat DEF": "防御力", + "ATK%": "百分比攻击力", + "HP%": "百分比血量", + "DEF%": "百分比防御", + "Elemental Mastery": "元素精通", + "Energy Recharge": "元素充能效率", + "Crit RATE": "暴击率", + "Crit DMG": "暴击伤害", + "Cryo DMG Bonus": "冰元素伤害加成", + "Pyro DMG Bonus": "火元素伤害加成", + "Hydro DMG Bonus": "水元素伤害加成", + "Electro DMG Bonus": "雷元素伤害加成", + "Anemo DMG Bonus": "风元素伤害加成", + "Geo DMG Bonus": "岩元素伤害加成", + "Dendro DMG Bonus": "草元素伤害加成", + "Healing Bonus": "治疗加成", + "Physical Bonus": "物理伤害加成", + } + + def __init__(self): + self.client = httpx.AsyncClient(timeout=60) + self.session_id = None + + async def get_session_id(self) -> Optional[str]: + if self.session_id is None: + resp = await self.client.get(MAIN_API) + sid = resp.cookies.get("connect.sid", "") + sid = unquote(str(sid)) + self.session_id = sid.split(".")[0].split(":")[-1] + return self.session_id + + async def refresh_user_data(self, uid: int) -> None: + session_id = await self.get_session_id() + params = {"sessionID": session_id} + await self.client.get(DATA_API + str(uid), params=params) + await self.client.get(REFRESH_API + str(uid), params=params) + + async def get_rank_data(self, uid: int) -> List[AkashaRank]: + await self.refresh_user_data(uid) + try: + resp = await self.client.get(RANK_API + str(uid)) + data = resp.json()["data"] + except KeyError: + return [] + return [AkashaRank(**i) for i in data] + + async def get_leaderboard_categories(self, character_id: int) -> List[AkashaLeaderboardCategory]: + params = {"characterId": character_id} + try: + resp = await self.client.get(LEADERBOARD_CATEGORY_API, params=params) + data = resp.json()["data"] + except KeyError: + return [] + return [AkashaLeaderboardCategory(**i) for i in data] + + async def get_leaderboard(self, calculation_id: str, uid: int = None) -> List[AkashaLeaderboard]: + params = { + "sort": "calculation.result", + "p": "", + "calculationId": calculation_id, + "order": -1, + "size": 20, + "page": 1, + "filter": "", + "uids": "", + "fromId": "", + } + if uid: + params["uids"] = f"[uid]{uid}" + try: + resp = await self.client.get(LEADERBOARD_API, params=params) + data = resp.json()["data"] + except KeyError: + return [] + return [AkashaLeaderboard(**i) for i in data] + + async def get_artifacts_list(self, sort_by: AkashaSubStat = AkashaSubStat.CRR) -> List[AkashaArtifact]: + params = { + "sort": self.SUB_STAT_MAP[sort_by], + "p": "", + } + try: + resp = await self.client.get(ARTIFACTS_API, params=params) + data = resp.json()["data"] + except KeyError: + return [] + return [AkashaArtifact(**i) for i in data] + + async def __aenter__(self): + return self + + async def __aexit__( + self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType] + ): + if self.client.is_closed: + return + await self.client.aclose() diff --git a/modules/apihelper/models/genshin/akasha.py b/modules/apihelper/models/genshin/akasha.py new file mode 100644 index 00000000..c4a50ece --- /dev/null +++ b/modules/apihelper/models/genshin/akasha.py @@ -0,0 +1,212 @@ +from datetime import datetime +from enum import Enum +from typing import Dict, List, Any, Optional + +from pydantic import BaseModel, Field + + +class AkashaSubStat(str, Enum): + CRR = "双暴" + ATK = "百分比攻击力" + HP = "百分比生命值" + DEF = "百分比防御力" + ATKF = "固定攻击力" + HPF = "固定生命值" + DEFF = "固定防御力" + EM = "元素精通" + ER = "元素充能效率" + CR = "暴击率" + CD = "暴击伤害" + + +class AkashaRankCalFit(BaseModel): + calculationId: str + short: str + name: str + details: str + result: float + ranking: str + outOf: int + priority: int + type: str + + +class AkashaRankCal(BaseModel): + fit: AkashaRankCalFit + + +class AkashaRank(BaseModel): + _id: str + characterId: int + uid = int + constellation: int + icon: str + + +class AkashaLeaderboardCategoryWeapon(BaseModel): + name: str + icon: str + substat: str + type: str + rarity: str + refinement: int + calculationId: str + details: str + + +class AkashaLeaderboardCategory(BaseModel): + _id: str + name: str + addDate: datetime + c6: str + characterId: int + characterName: str + count: int + details: str + element: str + new: int + rarity: int + short: str + weapons: List[AkashaLeaderboardCategoryWeapon] + weaponsCount: int + characterIcon: str + index: int + + +class AkashaLeaderboardCalculation(BaseModel): + id: str + result: float + + @property + def int(self) -> int: + return int(self.result) + + +class AkashaLeaderboardArtifactSet(BaseModel): + icon: str + count: int + + +class AkashaLeaderboardOwner(BaseModel): + nickname: str + adventureRank: float + profilePicture: Any + nameCard: str + patreon: Dict[str, Any] + region: str + + +class AkashaLeaderboardStatsValue(BaseModel): + value: float + + @property + def int(self) -> int: + return int(self.value) + + @property + def percent(self) -> str: + return f"{self.value * 100:.1f}" + + @property + def web_value(self) -> str: + return f"{self.value:.2f}" + + +class AkashaLeaderboardStats(BaseModel): + maxHp: AkashaLeaderboardStatsValue + atk: AkashaLeaderboardStatsValue + def_: AkashaLeaderboardStatsValue = Field(..., alias="def") + elementalMastery: AkashaLeaderboardStatsValue + energyRecharge: AkashaLeaderboardStatsValue + healingBonus: AkashaLeaderboardStatsValue + critRate: AkashaLeaderboardStatsValue + critDamage: AkashaLeaderboardStatsValue + electroDamageBonus: Optional[AkashaLeaderboardStatsValue] + + +class AkashaLeaderboardWeaponInfo(BaseModel): + level: int + promoteLevel: int + refinementLevel: AkashaLeaderboardStatsValue + + +class AkashaLeaderboardWeapon(BaseModel): + weaponInfo: AkashaLeaderboardWeaponInfo + flat: Dict[str, Any] + name: str + icon: str + + +class AkashaLeaderboardCharacterMetadata(BaseModel): + element: str + + +class AkashaLeaderboard(BaseModel): + _id: str + calculation: AkashaLeaderboardCalculation + characterId: int + type: str + uid: str + artifactObjects: Dict[str, Any] + artifactSets: Dict[str, AkashaLeaderboardArtifactSet] + calculations: Dict[str, Any] + constellation: int + costumeId: str + critValue: float + md5: str + name: str + owner: AkashaLeaderboardOwner + propMap: Dict[str, Any] + proudSkillExtraLevelMap: Dict[str, Any] + stats: AkashaLeaderboardStats + talentsLevelMap: Dict[str, Any] + weapon: AkashaLeaderboardWeapon + icon: str + index: str + nameCardLink: str + profilePictureLink: str + characterMetadata: AkashaLeaderboardCharacterMetadata + + +class AkashaArtifactType(str, Enum): + BRACER = "EQUIP_BRACER" + """生之花""" + NECKLACE = "EQUIP_NECKLACE" + """死之羽""" + SHOES = "EQUIP_SHOES" + """时之沙""" + RING = "EQUIP_RING" + """空之杯""" + DRESS = "EQUIP_DRESS" + """理之冠""" + + @property + def real_name(self): + name_map = { + "EQUIP_BRACER": "生之花", + "EQUIP_NECKLACE": "死之羽", + "EQUIP_SHOES": "时之沙", + "EQUIP_RING": "空之杯", + "EQUIP_DRESS": "理之冠", + } + return name_map[self.value] + + +class AkashaArtifact(BaseModel): + _id: str + uid: int + critValue: float + equipType: AkashaArtifactType + icon: str + level: int + mainStatKey: str + mainStatValue: float + name: str + owner: AkashaLeaderboardOwner + setName: str + stars: int + substats: Dict[str, float] + substatsIdList: List[int] + index: int + nameCardLink: str + profilePictureLink: str diff --git a/plugins/genshin/akasha.py b/plugins/genshin/akasha.py new file mode 100644 index 00000000..115316a7 --- /dev/null +++ b/plugins/genshin/akasha.py @@ -0,0 +1,112 @@ +from typing import TYPE_CHECKING, Optional + +from telegram.constants import ChatAction +from telegram.ext import filters + +from core.dependence.assets import AssetsService +from core.plugin import Plugin, handler +from core.services.template.models import FileType +from core.services.template.services import TemplateService +from gram_core.services.players import PlayersService +from metadata.genshin import AVATAR_DATA +from metadata.shortname import roleToName, roleToId +from modules.apihelper.client.components.akasha import Akasha +from utils.log import logger + +if TYPE_CHECKING: + from telegram import Update + from telegram.ext import ContextTypes + + +class AkashaPlugin(Plugin): + """Akasha 数据排行""" + + def __init__( + self, + assets_service: AssetsService = None, + template_service: TemplateService = None, + player_service: PlayersService = None, + ) -> None: + self.assets_service = assets_service + self.template_service = template_service + self.player_service = player_service + + async def get_user_uid(self, user_id: int) -> Optional[int]: + player = await self.player_service.get(user_id) + if player is None: + return None + return player.player_id + + @staticmethod + async def get_leaderboard_data(character_id: int, uid: int = None): + akasha = Akasha() + categories = await akasha.get_leaderboard_categories(character_id) + if len(categories) == 0 or len(categories[0].weapons) == 0: + raise NotImplementedError + calculation_id = categories[0].weapons[0].calculationId + count = categories[0].count + data = await akasha.get_leaderboard(calculation_id) + if len(data) == 0: + raise NotImplementedError + user_data = [] + if uid: + user_data = await akasha.get_leaderboard(calculation_id, uid) + if len(user_data) == 0: + data = [data] + else: + data = [user_data, data] + return data, count + + async def get_avatar_board_render_data(self, character: str, uid: int): + character_id = roleToId(character) + name_card = (await self.assets_service.namecard(character_id).navbar()).as_uri() + avatar = (await self.assets_service.avatar(character_id).icon()).as_uri() + rarity = 5 + try: + rarity = {k: v["rank"] for k, v in AVATAR_DATA.items()}[str(character_id)] + except KeyError: + logger.warning("未找到角色 %s 的星级", character_id) + akasha_data, count = await self.get_leaderboard_data(character_id, uid) + return { + "character": character, # 角色名 + "avatar": avatar, # 角色头像 + "namecard": name_card, # 角色名片 + "rarity": rarity, # 角色稀有度 + "count": count, + "all_data": akasha_data, + } + + @handler.command("avatar_board", block=False) + @handler.message(filters.Regex(r"^角色排名(.*)$"), block=False) + async def avatar_board(self, update: "Update", context: "ContextTypes.DEFAULT_TYPE"): + user = update.effective_user + message = update.effective_message + args = self.get_args(context) + if len(args) == 0: + reply_message = await message.reply_text("请指定要查询的角色") + if filters.ChatType.GROUPS.filter(reply_message): + self.add_delete_message_job(message) + self.add_delete_message_job(reply_message) + return + avatar_name = roleToName(args[0]) + uid = await self.get_user_uid(user.id) + try: + render_data = await self.get_avatar_board_render_data(avatar_name, uid) + except NotImplementedError: + reply_message = await message.reply_text("暂不支持该角色") + if filters.ChatType.GROUPS.filter(reply_message): + self.add_delete_message_job(message) + self.add_delete_message_job(reply_message) + return + await message.reply_chat_action(ChatAction.UPLOAD_PHOTO) + + image = await self.template_service.render( + "genshin/akasha/char_rank.jinja2", + render_data, + viewport={"width": 1040, "height": 500}, + full_page=True, + query_selector=".container", + file_type=FileType.PHOTO, + ttl=24 * 60 * 60, + ) + await image.reply_photo(message) diff --git a/resources/genshin/akasha/char_rank.jinja2 b/resources/genshin/akasha/char_rank.jinja2 new file mode 100644 index 00000000..3fd0db72 --- /dev/null +++ b/resources/genshin/akasha/char_rank.jinja2 @@ -0,0 +1,72 @@ + + +
+ +