diff --git a/modules/apihelper/abyss_team.py b/modules/apihelper/abyss_team.py new file mode 100644 index 0000000..1512307 --- /dev/null +++ b/modules/apihelper/abyss_team.py @@ -0,0 +1,68 @@ +import time +from typing import List, Optional + +import httpx +from pydantic import BaseModel, validator, parse_obj_as + + +class Member(BaseModel): + star: int + attr: str + name: str + + +class TeamRate(BaseModel): + rate: float + formation: List[Member] + ownerNum: Optional[int] + + @validator('rate', pre=True) + def str2float(cls, v): # pylint: disable=R0201 + return float(v.replace('%', '')) / 100.0 if isinstance(v, str) else v + + +class TeamRateResult(BaseModel): + rateListUp: List[TeamRate] + rateListDown: List[TeamRate] + userCount: int + + def sort(self, characters: List[str]): + for team in self.rateListUp: + team.ownerNum = len(set(characters) & {member.name for member in team.formation}) + for team in self.rateListDown: + team.ownerNum = len(set(characters) & {member.name for member in team.formation}) + self.rateListUp.sort(key=lambda x: (x.ownerNum / 4 * x.rate), reverse=True) + self.rateListDown.sort(key=lambda x: (x.ownerNum / 4 * x.rate), reverse=True) + + +class AbyssTeamData: + TEAM_RATE_API = "https://www.youchuang.fun/gamerole/formationRate" + HEADERS = { + 'Host': 'www.youchuang.fun', + 'Referer': 'https://servicewechat.com/wxce4dbe0cb0f764b3/91/page-frame.html', + 'User-Agent': 'Mozilla/5.0 (iPad; CPU OS 15_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) ' + 'Mobile/15E148 MicroMessenger/8.0.20(0x1800142f) NetType/WIFI Language/zh_CN', + 'content-type': 'application/json' + } + VERSION = "3.1" + + def __init__(self): + self.client = httpx.AsyncClient(headers=self.HEADERS) + self.time = 0 + self.data = None + self.ttl = 10 * 60 + + async def get_data(self) -> TeamRateResult: + if self.data is None or self.time + self.ttl < time.time(): + data_up = await self.client.post(self.TEAM_RATE_API, json={"version": self.VERSION, "layer": 1}) + data_up_json = data_up.json()["result"] + data_down = await self.client.post(self.TEAM_RATE_API, json={"version": self.VERSION, "layer": 2}) + data_down_json = data_down.json()["result"] + self.data = TeamRateResult(rateListUp=parse_obj_as(List[TeamRate], data_up_json["rateList"]), + rateListDown=parse_obj_as(List[TeamRate], data_down_json["rateList"]), + userCount=data_up_json["userCount"]) + self.time = time.time() + return self.data.copy(deep=True) + + async def close(self): + await self.client.aclose() diff --git a/plugins/genshin/abyss_team.py b/plugins/genshin/abyss_team.py new file mode 100644 index 0000000..e9fe434 --- /dev/null +++ b/plugins/genshin/abyss_team.py @@ -0,0 +1,88 @@ +from telegram import Update, User +from telegram.constants import ChatAction +from telegram.ext import CallbackContext, CommandHandler, MessageHandler, filters + +from core.assets import AssetsService +from core.baseplugin import BasePlugin +from core.cookies.error import CookiesNotFoundError +from core.plugin import Plugin, handler +from core.template import TemplateService +from core.user import UserService +from core.user.error import UserNotFoundError +from metadata.shortname import roleToId +from modules.apihelper.abyss_team import AbyssTeamData +from utils.decorators.error import error_callable +from utils.decorators.restricts import restricts +from utils.helpers import get_genshin_client +from utils.log import logger + + +class AbyssTeam(Plugin, BasePlugin): + """深境螺旋推荐配队查询""" + + def __init__(self, user_service: UserService = None, template_service: TemplateService = None, + assets: AssetsService = None): + self.template_service = template_service + self.user_service = user_service + self.assets_service = assets + self.team_data = AbyssTeamData() + + @staticmethod + def _get_role_star_bg(value: int): + if value == 4: + return "./../abyss/background/roleStarBg4.png" + elif value == 5: + return "./../abyss/background/roleStarBg5.png" + else: + raise ValueError("错误的数据") + + @staticmethod + async def _get_data_from_user(user: User): + try: + logger.debug("尝试获取已绑定的原神账号") + client = await get_genshin_client(user.id) + logger.debug(f"获取成功, UID: {client.uid}") + characters = await client.get_genshin_characters(client.uid) + return [character.name for character in characters] + except (UserNotFoundError, CookiesNotFoundError): + logger.info(f"未查询到用户({user.full_name} {user.id}) 所绑定的账号信息") + return [] + + @handler(CommandHandler, command="abyss_team", block=False) + @handler(MessageHandler, filters=filters.Regex("^深渊推荐配队(.*)"), block=False) + @restricts() + @error_callable + async def command_start(self, update: Update, _: CallbackContext) -> None: + user = update.effective_user + message = update.effective_message + logger.info(f"用户 {user.full_name}[{user.id}] 查深渊推荐配队命令请求") + await message.reply_chat_action(ChatAction.TYPING) + team_data = await self.team_data.get_data() + # 尝试获取用户已绑定的原神账号信息 + user_data = await self._get_data_from_user(user) + team_data.sort(user_data) + abyss_team_data = { + "up": [], + "down": [] + } + for i in team_data.rateListUp[0].formation: + temp = { + "icon": (await self.assets_service.character(roleToId(i.name)).icon()).as_uri(), + "name": i.name, + "background": self._get_role_star_bg(i.star), + "hava": i.name in user_data, + } + abyss_team_data["up"].append(temp) + for i in team_data.rateListDown[0].formation: + temp = { + "icon": (await self.assets_service.character(roleToId(i.name)).icon()).as_uri(), + "name": i.name, + "background": self._get_role_star_bg(i.star), + "hava": i.name in user_data, + } + abyss_team_data["down"].append(temp) + await message.reply_chat_action(ChatAction.UPLOAD_PHOTO) + png_data = await self.template_service.render('genshin/abyss_team', "abyss_team.html", abyss_team_data, + {"width": 865, "height": 504}, full_page=False) + await message.reply_photo(png_data, filename=f"abyss_team_{user.id}.png", + allow_sending_without_reply=True) diff --git a/resources/bot/help/help.html b/resources/bot/help/help.html index 27d7c75..4b09f97 100644 --- a/resources/bot/help/help.html +++ b/resources/bot/help/help.html @@ -90,6 +90,11 @@
/material
角色培养素材查询
+
+
/abyss_team
+
深渊推荐配队
+
+
/hilichurls
丘丘语字典
diff --git a/resources/genshin/abyss_team/abyss_team.html b/resources/genshin/abyss_team/abyss_team.html new file mode 100644 index 0000000..a5951f9 --- /dev/null +++ b/resources/genshin/abyss_team/abyss_team.html @@ -0,0 +1,69 @@ + + + + + abyss + + + + + +
+
深境螺旋 - 推荐配队
+
+
第 12 层上半
+
+ {% for i in up %} +
+
+
+
{{i.name}}
+
+ {% endfor %} +
+
+
+
第 12 层下半
+
+ {% for i in down %} +
+
+
+
{{i.name}}
+
+ {% endfor %} +
+
+
+
+ + \ No newline at end of file diff --git a/resources/genshin/abyss_team/example.html b/resources/genshin/abyss_team/example.html new file mode 100644 index 0000000..02cdcff --- /dev/null +++ b/resources/genshin/abyss_team/example.html @@ -0,0 +1,100 @@ + + + + + abyss + + + + +
+
深境螺旋 - 推荐配队
+
+
第 12 层上半
+
+
+
+
+
xx
+
+
+
+
+
xx
+
+
+
+
+
xx
+
+
+
+
+
xx
+
+
+
+
+
第 12 层下半
+
+
+
+
+
xx
+
+
+
+
+
xx
+
+
+
+
+
xx
+
+
+
+
+
xx
+
+
+
+
+
+ + \ No newline at end of file diff --git a/tests/test_abyss_team_data.py b/tests/test_abyss_team_data.py new file mode 100644 index 0000000..93564ed --- /dev/null +++ b/tests/test_abyss_team_data.py @@ -0,0 +1,33 @@ +import logging + +import pytest +import pytest_asyncio +from flaky import flaky + +from modules.apihelper.abyss_team import AbyssTeamData, TeamRateResult, TeamRate + +LOGGER = logging.getLogger(__name__) + + +@pytest_asyncio.fixture +async def abyss_team_data(): + _abyss_team_data = AbyssTeamData() + yield _abyss_team_data + await _abyss_team_data.close() + + +# noinspection PyShadowingNames +@pytest.mark.asyncio +@flaky(3, 1) +async def test_abyss_team_data(abyss_team_data: AbyssTeamData): + team_data = await abyss_team_data.get_data() + assert isinstance(team_data, TeamRateResult) + assert isinstance(team_data.rateListUp[0], TeamRate) + assert isinstance(team_data.rateListUp[-1], TeamRate) + assert isinstance(team_data.rateListDown[0], TeamRate) + assert isinstance(team_data.rateListDown[-1], TeamRate) + assert team_data.userCount > 0 + for i in team_data.rateListUp[0].formation: + LOGGER.info("rate down info:name %s star %s", i.name, i.star) + for i in team_data.rateListDown[0].formation: + LOGGER.info("rate up info:name %s star %s", i.name, i.star)