"""Recommend teams for Spiral Abyss""" import re from telegram import Update from telegram.constants import ChatAction, ParseMode from telegram.ext import CallbackContext, filters from core.dependence.assets import AssetsService from core.plugin import Plugin, handler from core.services.template.services import TemplateService from metadata.genshin import AVATAR_DATA from metadata.shortname import idToName from modules.apihelper.client.components.abyss import AbyssTeam as AbyssTeamClient from plugins.tools.genshin import GenshinHelper from utils.log import logger class AbyssTeamPlugin(Plugin): """Recommend teams for Spiral Abyss""" def __init__( self, template: TemplateService, helper: GenshinHelper, assets_service: AssetsService, ): self.template_service = template self.helper = helper self.team_data = AbyssTeamClient() self.assets_service = assets_service @handler.command("abyss_team", block=False) @handler.message(filters.Regex(r"^深渊配队"), block=False) async def command_start(self, update: Update, _: CallbackContext) -> None: # skipcq: PY-R1000 # user_id = await self.get_real_user_id(update) message = update.effective_message if "help" in message.text or "帮助" in message.text: await message.reply_text( "深渊配队推荐功能使用帮助(中括号表示可选参数)\n\n" "指令格式:\n/abyss_team [n=配队数]\n(pre表示上期)\n\n" "文本格式:\n深渊配队 [n=配队数] \n\n" "如:\n" "/abyss_team\n/abyss_team n=5\n" "深渊配队\n", parse_mode=ParseMode.HTML, ) self.log_user(update, logger.info, "查询[bold]深渊配队推荐[/bold]帮助", extra={"markup": True}) return self.log_user(update, logger.info, "[bold]深渊配队推荐[/bold]请求", extra={"markup": True}) client = await self.helper.get_genshin_client(user_id) await message.reply_chat_action(ChatAction.TYPING) team_data = await self.team_data.get_data() # Set of uids characters = {c.id for c in await client.get_genshin_characters(client.player_id)} teams = { "Up": [], "Down": [], } # All of the effective and available teams for lane in ["Up", "Down"]: for a_team in team_data[12 - 9][lane]: t_characters = [int(s) for s in re.findall(r"\d+", a_team["Item"])] t_rate = a_team["Rate"] # Check availability if not all(c in characters for c in t_characters): continue teams[lane].append( { "Characters": t_characters, "Rate": t_rate, } ) # If a number is specified, use it as the number of expected teams. match = re.search(r"(?<=n=)\d+", message.text) n_team = int(match.group()) if match is not None else 4 if "fast" in message.text: # TODO: Give it a faster method? # Maybe we can allow characters exist on both side. return # Otherwise, we'd find a team in a complexity # O(len(teams[up]) * len(teams[down])) abyss_teams_data = {"uid": client.player_id, "teams": []} async def _get_render_data(id_list): return [ { "icon": (await self.assets_service.avatar(cid).icon()).as_uri(), "name": idToName(cid), "star": AVATAR_DATA[str(cid)]["rank"] if cid not in {10000005, 10000007} else 5, "hava": True, } for cid in id_list ] for u in teams["Up"]: for d in teams["Down"]: if not all(c not in d["Characters"] for c in u["Characters"]): continue team = { "Up": await _get_render_data(u["Characters"]), "UpRate": u["Rate"], "Down": await _get_render_data(d["Characters"]), "DownRate": d["Rate"], } abyss_teams_data["teams"].append(team) abyss_teams_data["teams"].sort(key=lambda t: t["UpRate"] * t["DownRate"], reverse=True) abyss_teams_data["teams"] = abyss_teams_data["teams"][0 : min(n_team, len(abyss_teams_data["teams"]))] await message.reply_chat_action(ChatAction.UPLOAD_PHOTO) render_result = await self.template_service.render( "genshin/abyss_team/abyss_team.jinja2", abyss_teams_data, {"width": 785, "height": 800}, full_page=True, query_selector=".bg-contain", ) await render_result.reply_photo(message, filename=f"abyss_team_{user_id}.png", allow_sending_without_reply=True)