mirror of
https://github.com/RYGhub/royalnet.git
synced 2024-12-17 23:24:20 +00:00
221 lines
10 KiB
Python
221 lines
10 KiB
Python
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":
|
||
self.loop.create_task(self._updater(900))
|
||
|
||
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)
|