mirror of
https://github.com/PaiGramTeam/PamGram.git
synced 2024-11-22 14:26:45 +00:00
230 lines
7.5 KiB
Python
230 lines
7.5 KiB
Python
from decimal import Decimal
|
|
from typing import List, Optional, Union, Dict
|
|
|
|
import ujson
|
|
from httpx import AsyncClient, TimeoutException
|
|
from pydantic import BaseModel
|
|
|
|
from core.config import config
|
|
from modules.playercards.fight_prop import EquipmentsStats
|
|
from modules.wiki.base import WikiModel
|
|
from modules.wiki.models.relic_affix import RelicAffixAll
|
|
from utils.enkanetwork import RedisCache
|
|
from modules.playercards.file import PlayerCardsFile
|
|
|
|
|
|
class SkillTreePoint(BaseModel):
|
|
pointId: int
|
|
level: int
|
|
|
|
|
|
class Equipment(BaseModel):
|
|
tid: Optional[int] = 0
|
|
level: Optional[int] = 0
|
|
promotion: Optional[int] = 3
|
|
"""星级"""
|
|
rank: Optional[int] = 0
|
|
"""叠影"""
|
|
|
|
|
|
class SubAffix(BaseModel):
|
|
cnt: Optional[int] = 1
|
|
step: Optional[int] = 0
|
|
affixId: int
|
|
|
|
|
|
class Relic(BaseModel):
|
|
tid: int
|
|
level: Optional[int] = 0
|
|
mainAffixId: int
|
|
subAffixList: Optional[List[SubAffix]]
|
|
type: int
|
|
|
|
|
|
class Property(BaseModel):
|
|
name: str
|
|
base: float = 0.0
|
|
addition: float = 0.0
|
|
percent: bool
|
|
|
|
@property
|
|
def total(self):
|
|
total_num = (Decimal(self.base) + Decimal(self.addition)) * (Decimal(100.0) if self.percent else Decimal(1.0))
|
|
total_num = round(total_num, 2)
|
|
return f"{total_num}{'%' if self.percent else ''}"
|
|
|
|
|
|
class Avatar(BaseModel):
|
|
avatarId: int
|
|
skillTreeList: List[SkillTreePoint]
|
|
equipment: Optional[Equipment]
|
|
level: int
|
|
promotion: Optional[int] = 4
|
|
rank: Optional[int] = 0
|
|
relicList: Optional[List[Relic]]
|
|
property: Optional[List[Property]]
|
|
|
|
|
|
class RecordInfo(BaseModel):
|
|
achievementCount: Optional[int] = 0
|
|
avatarCount: Optional[int] = 0
|
|
equipmentCount: Optional[int] = 0
|
|
maxRogueChallengeScore: Optional[int] = 0
|
|
|
|
|
|
class PlayerBaseInfo(BaseModel):
|
|
platform: Optional[str]
|
|
friendCount: Optional[int]
|
|
headIcon: Optional[int]
|
|
isDisplayAvatar: bool
|
|
level: int
|
|
worldLevel: Optional[int]
|
|
nickname: str
|
|
recordInfo: RecordInfo
|
|
signature: Optional[str]
|
|
uid: int
|
|
|
|
|
|
class PlayerInfo(PlayerBaseInfo):
|
|
avatarList: List[Avatar]
|
|
|
|
|
|
class PlayerCardsError(Exception):
|
|
def __init__(self, msg):
|
|
self.msg = msg
|
|
|
|
|
|
class PlayerCards:
|
|
url = "https://api.mihomo.me/sr_info/"
|
|
url2 = "https://api.mihomo.me/sr_info_parsed/"
|
|
prop_url = f"{WikiModel.BASE_URL}relic_config.json"
|
|
|
|
def __init__(self, redis):
|
|
self.cache = RedisCache(redis.client, key="plugin:player_cards:fake_enka_network", ex=60)
|
|
self.headers = {"User-Agent": config.enka_network_api_agent}
|
|
self.client = AsyncClient()
|
|
self.player_cards_file = PlayerCardsFile()
|
|
self.init = False
|
|
self.relic_datas_map: Dict[int, RelicAffixAll] = {}
|
|
|
|
async def async_init(self):
|
|
if self.init:
|
|
return
|
|
self.relic_datas_map.clear()
|
|
req = await self.client.get(self.prop_url)
|
|
data = req.json()
|
|
for i in data:
|
|
self.relic_datas_map[i["id"]] = RelicAffixAll(**i)
|
|
self.init = True
|
|
|
|
async def get_property(self, uid: str) -> Dict[int, List[Dict]]:
|
|
final_data: Dict[int, List[Dict]] = {}
|
|
try:
|
|
user = await self.client.get(self.url2 + uid, timeout=30, headers=self.headers)
|
|
if user.status_code != 200:
|
|
raise PlayerCardsError("请求异常")
|
|
data = ujson.loads(user.text)
|
|
characters = data.get("characters", [])
|
|
for character in characters:
|
|
cid = int(character.get("id", 0))
|
|
if not cid:
|
|
continue
|
|
datas = []
|
|
datas_map = {}
|
|
for attr in character.get("attributes", []):
|
|
prop = Property(
|
|
name=attr["name"],
|
|
base=attr["value"],
|
|
percent=attr["percent"],
|
|
)
|
|
datas.append(prop)
|
|
datas_map[prop.name] = prop
|
|
for attr in character.get("additions", []):
|
|
prop = datas_map.get(attr["name"])
|
|
if prop:
|
|
prop.addition = attr["value"]
|
|
else:
|
|
prop = Property(
|
|
name=attr["name"],
|
|
addition=attr["value"],
|
|
percent=attr["percent"],
|
|
)
|
|
datas.append(prop)
|
|
datas_map[prop.name] = prop
|
|
final_data[cid] = [i.dict() for i in datas]
|
|
except (TimeoutException, PlayerCardsError):
|
|
pass
|
|
return final_data
|
|
|
|
async def update_data(self, uid: str) -> Union[PlayerInfo, str]:
|
|
try:
|
|
data = await self.cache.get(uid)
|
|
if data is not None:
|
|
return PlayerInfo.parse_obj(data)
|
|
user = await self.client.get(self.url + uid, timeout=30, headers=self.headers)
|
|
if user.status_code != 200:
|
|
raise PlayerCardsError(f"请求异常，错误代码 {user.status_code}")
|
|
data = ujson.loads(user.text)
|
|
error_code = data.get("ErrCode", 0)
|
|
if error_code:
|
|
raise PlayerCardsError(f"请求异常，错误代码 {error_code}")
|
|
data = data.get("detailInfo", {})
|
|
props = await self.get_property(uid)
|
|
data = await self.player_cards_file.merge_info(uid, data, props)
|
|
await self.cache.set(uid, data)
|
|
return PlayerInfo.parse_obj(data)
|
|
except TimeoutException:
|
|
error = "服务请求超时，请稍后重试"
|
|
except PlayerCardsError as e:
|
|
error = e.msg
|
|
return error
|
|
|
|
async def get_player_base_info(self, uid: int) -> PlayerBaseInfo:
|
|
try:
|
|
user = await self.client.get(f"{self.url}{uid}", timeout=30, headers=self.headers)
|
|
if user.status_code != 200:
|
|
raise PlayerCardsError(f"请求异常，错误代码 {user.status_code}")
|
|
data = ujson.loads(user.text)
|
|
error_code = data.get("ErrCode", 0)
|
|
if error_code:
|
|
raise PlayerCardsError(f"请求异常，错误代码 {error_code}")
|
|
return PlayerBaseInfo.parse_obj(data["detailInfo"])
|
|
except TimeoutException as e:
|
|
raise PlayerCardsError("服务请求超时，请稍后重试") from e
|
|
|
|
def get_affix_by_id(self, cid: int) -> RelicAffixAll:
|
|
return self.relic_datas_map.get(cid)
|
|
|
|
def get_set_by_id(self, cid: int) -> int:
|
|
if affix := self.get_affix_by_id(cid):
|
|
return affix.set_id
|
|
return 101
|
|
|
|
def get_affix(self, relic: Relic, main: bool = True, sub: bool = True) -> List[EquipmentsStats]:
|
|
affix = self.get_affix_by_id(relic.tid)
|
|
if not affix:
|
|
return []
|
|
main_affix = affix.main_affix[str(relic.mainAffixId)]
|
|
datas = (
|
|
[
|
|
EquipmentsStats(
|
|
prop_id=main_affix.property,
|
|
prop_value=main_affix.get_value(relic.level),
|
|
)
|
|
]
|
|
if main
|
|
else []
|
|
)
|
|
if not sub:
|
|
return datas
|
|
if relic.subAffixList:
|
|
for sub_a in relic.subAffixList:
|
|
sub_affix = affix.sub_affix[str(sub_a.affixId)]
|
|
datas.append(
|
|
EquipmentsStats(
|
|
prop_id=sub_affix.property,
|
|
prop_value=sub_affix.get_value(sub_a.step, sub_a.cnt),
|
|
)
|
|
)
|
|
return datas
|