import typing import riotwatcher import logging import asyncio import sentry_sdk from royalnet.commands import * from royalnet.utils import * from royalnet.serf.telegram import * from ..tables import LeagueOfLegends from ..types import LeagueLeague log = logging.getLogger(__name__) class LeagueoflegendsCommand(Command): name: str = "leagueoflegends" aliases = ["lol", "league"] description: str = "Connetti un account di League of Legends a un account Royalnet, e visualizzane le statistiche." syntax = "[nomeevocatore]" def __init__(self, interface: CommandInterface): super().__init__(interface) self._riotwatcher = riotwatcher.RiotWatcher(api_key=self.config["Lol"]["token"]) if self.interface.name == "telegram" and self.config["Lol"]["updater"]: self.loop.create_task(self._updater(7200)) async def _send(self, message): client = self.serf.client await self.serf.api_call(client.send_message, chat_id=self.config["Telegram"]["main_group_id"], text=escape(message), parse_mode="HTML", disable_webpage_preview=True) async def _notify(self, obj: LeagueOfLegends, attribute_name: str, old_value: typing.Any, new_value: typing.Any): if isinstance(old_value, LeagueLeague): # This is a rank change! # Don't send messages for every rank change, send messages just if the TIER or RANK changes! if old_value.tier == new_value.tier and old_value.rank == new_value.rank: return # Find the queue queue_names = { "rank_soloq": "Solo/Duo", "rank_flexq": "Flex", "rank_twtrq": "3v3", "rank_tftq": "TFT" } # Prepare the message if new_value > old_value: message = f"📈 [b]{obj.user}[/b] è salito a {new_value} su League of Legends " \ f"({queue_names[attribute_name]})! Congratulazioni!" else: message = f"📉 [b]{obj.user}[/b] è sceso a {new_value} su League of Legends " \ f"({queue_names[attribute_name]})." # Send the message await self._send(message) # Level up! elif attribute_name == "summoner_level": if new_value == 30 or (new_value >= 50 and (new_value % 25 == 0)): await self._send(f"🆙 [b]{obj.user}[/b] è salito al livello [b]{new_value}[/b] su League of Legends!") @staticmethod async def _change(obj: LeagueOfLegends, attribute_name: str, new_value: typing.Any, callback: typing.Callable[ [LeagueOfLegends, str, typing.Any, typing.Any], typing.Awaitable[None]]): old_value = obj.__getattribute__(attribute_name) if old_value != new_value: await callback(obj, attribute_name, old_value, new_value) obj.__setattr__(attribute_name, new_value) async def _update(self, lol: LeagueOfLegends): log.info(f"Updating: {lol}") log.debug(f"Getting summoner data: {lol}") summoner = await asyncify(self._riotwatcher.summoner.by_id, region=self.config["Lol"]["region"], encrypted_summoner_id=lol.summoner_id) await self._change(lol, "profile_icon_id", summoner["profileIconId"], self._notify) await self._change(lol, "summoner_name", summoner["name"], self._notify) await self._change(lol, "puuid", summoner["puuid"], self._notify) await self._change(lol, "summoner_level", summoner["summonerLevel"], self._notify) await self._change(lol, "summoner_id", summoner["id"], self._notify) await self._change(lol, "account_id", summoner["accountId"], self._notify) log.debug(f"Getting leagues data: {lol}") leagues = await asyncify(self._riotwatcher.league.by_summoner, region=self.config["Lol"]["region"], encrypted_summoner_id=lol.summoner_id) soloq = LeagueLeague() flexq = LeagueLeague() twtrq = LeagueLeague() tftq = LeagueLeague() for league in leagues: if league["queueType"] == "RANKED_SOLO_5x5": soloq = LeagueLeague.from_dict(league) if league["queueType"] == "RANKED_FLEX_SR": flexq = LeagueLeague.from_dict(league) if league["queueType"] == "RANKED_FLEX_TT": twtrq = LeagueLeague.from_dict(league) if league["queueType"] == "RANKED_TFT": tftq = LeagueLeague.from_dict(league) await self._change(lol, "rank_soloq", soloq, self._notify) await self._change(lol, "rank_flexq", flexq, self._notify) await self._change(lol, "rank_twtrq", twtrq, self._notify) await self._change(lol, "rank_tftq", tftq, self._notify) log.debug(f"Getting mastery data: {lol}") mastery = await asyncify(self._riotwatcher.champion_mastery.scores_by_summoner, region=self.config["Lol"]["region"], encrypted_summoner_id=lol.summoner_id) await self._change(lol, "mastery_score", mastery, self._notify) async def _updater(self, period: int): log.info(f"Started updater with {period}s period") while True: log.info(f"Updating...") session = self.alchemy.Session() log.info("") lols = session.query(self.alchemy.get(LeagueOfLegends)).all() for lol in lols: try: await self._update(lol) except Exception as e: sentry_sdk.capture_exception(e) log.error(f"Error while updating {lol.user.username}: {e}") await asyncio.sleep(1) await asyncify(session.commit) session.close() log.info(f"Sleeping for {period}s") await asyncio.sleep(period) @staticmethod def _display(lol: LeagueOfLegends) -> str: string = f"ℹ️ [b]{lol.summoner_name}[/b]\n" \ f"Lv. {lol.summoner_level}\n" \ f"Mastery score: {lol.mastery_score}\n" \ f"\n" if lol.rank_soloq: string += f"Solo: {lol.rank_soloq}\n" if lol.rank_flexq: string += f"Flex: {lol.rank_flexq}\n" if lol.rank_twtrq: string += f"3v3: {lol.rank_twtrq}\n" if lol.rank_tftq: string += f"TFT: {lol.rank_tftq}\n" return string async def run(self, args: CommandArgs, data: CommandData) -> None: author = await data.get_author(error_if_none=True) name = args.joined() if name: # Connect a new League of Legends account to Royalnet log.debug(f"Searching for: {name}") summoner = self._riotwatcher.summoner.by_name(region=self.config["Lol"]["region"], summoner_name=name) # Ensure the account isn't already connected to something else leagueoflegends = await asyncify( data.session.query(self.alchemy.get(LeagueOfLegends)).filter_by(summoner_id=summoner["id"]).one_or_none) if leagueoflegends: raise CommandError(f"L'account {leagueoflegends} è già registrato su Royalnet.") # Get rank information log.debug(f"Getting leagues data: {name}") leagues = self._riotwatcher.league.by_summoner(region=self.config["Lol"]["region"], encrypted_summoner_id=summoner["id"]) soloq = LeagueLeague() flexq = LeagueLeague() twtrq = LeagueLeague() tftq = LeagueLeague() for league in leagues: if league["queueType"] == "RANKED_SOLO_5x5": soloq = LeagueLeague.from_dict(league) if league["queueType"] == "RANKED_FLEX_SR": flexq = LeagueLeague.from_dict(league) if league["queueType"] == "RANKED_FLEX_TT": twtrq = LeagueLeague.from_dict(league) if league["queueType"] == "RANKED_TFT": tftq = LeagueLeague.from_dict(league) # Get mastery score log.debug(f"Getting mastery data: {name}") mastery = self._riotwatcher.champion_mastery.scores_by_summoner(region=self.config["Lol"]["region"], encrypted_summoner_id=summoner["id"]) # Create database row leagueoflegends = self.alchemy.get(LeagueOfLegends)( region=self.config["Lol"]["region"], user=author, profile_icon_id=summoner["profileIconId"], summoner_name=summoner["name"], puuid=summoner["puuid"], summoner_level=summoner["summonerLevel"], summoner_id=summoner["id"], account_id=summoner["accountId"], rank_soloq=soloq, rank_flexq=flexq, rank_twtrq=twtrq, rank_tftq=tftq, mastery_score=mastery ) log.debug(f"Saving to the DB: {name}") data.session.add(leagueoflegends) await data.session_commit() await data.reply(f"↔️ Account {leagueoflegends} connesso a {author}!") else: # Update and display the League of Legends stats for the current account if len(author.leagueoflegends) == 0: raise UserError("Nessun account di League of Legends trovato.") message = "" for account in author.leagueoflegends: try: await self._update(account) message += self._display(account) except riotwatcher.ApiError as e: message += f"⚠️ [b]{account.summoner_name}[/b]\n" \ f"{e}" message += "\n" await data.session_commit() await data.reply(message)