mirror of
https://github.com/RYGhub/royalnet.git
synced 2024-11-24 03:54:20 +00:00
171 lines
7.7 KiB
Python
171 lines
7.7 KiB
Python
from typing import *
|
||
import riotwatcher
|
||
import logging
|
||
import asyncio
|
||
import sentry_sdk
|
||
import royalnet.commands as rc
|
||
import royalnet.utils as ru
|
||
import royalnet.serf.telegram as rst
|
||
from royalnet.backpack import tables as rbt
|
||
|
||
from .abstract.linker import LinkerCommand
|
||
from ..tables import LeagueOfLegends, FiorygiTransaction
|
||
from ..types import LeagueLeague, Updatable
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
class LeagueoflegendsCommand(LinkerCommand):
|
||
name: str = "leagueoflegends"
|
||
|
||
aliases = ["lol", "league"]
|
||
|
||
description: str = "Connetti un account di League of Legends a un account Royalnet, o visualizzane le statistiche."
|
||
|
||
syntax = "[nomeevocatore]"
|
||
|
||
queue_names = {
|
||
"rank_soloq": "Solo/Duo",
|
||
"rank_flexq": "Flex",
|
||
"rank_twtrq": "3v3",
|
||
"rank_tftq": "TFT"
|
||
}
|
||
|
||
def __init__(self, interface: rc.CommandInterface):
|
||
super().__init__(interface)
|
||
self._riotwatcher: Optional[riotwatcher.RiotWatcher] = None
|
||
if self.interface.name == "telegram" and self.enabled():
|
||
self._riotwatcher = riotwatcher.RiotWatcher(api_key=self.token())
|
||
|
||
def token(self):
|
||
return self.config["leagueoflegends"]["token"]
|
||
|
||
def region(self):
|
||
return self.config["leagueoflegends"]["region"]
|
||
|
||
def describe(self, obj: LeagueOfLegends) -> str:
|
||
string = f"ℹ️ [b]{obj.summoner_name}[/b]\n" \
|
||
f"Lv. {obj.summoner_level}\n" \
|
||
f"Mastery score: {obj.mastery_score}\n" \
|
||
f"\n"
|
||
if obj.rank_soloq:
|
||
string += f"Solo: {obj.rank_soloq}\n"
|
||
if obj.rank_flexq:
|
||
string += f"Flex: {obj.rank_flexq}\n"
|
||
if obj.rank_twtrq:
|
||
string += f"3v3: {obj.rank_twtrq}\n"
|
||
if obj.rank_tftq:
|
||
string += f"TFT: {obj.rank_tftq}\n"
|
||
return string
|
||
|
||
async def get_updatables_of_user(self, session, user: rbt.User) -> List[LeagueOfLegends]:
|
||
return await ru.asyncify(session.query(self.alchemy.get(LeagueOfLegends)).filter_by(user=user).all)
|
||
|
||
async def get_updatables(self, session) -> List[LeagueOfLegends]:
|
||
return await ru.asyncify(session.query(self.alchemy.get(LeagueOfLegends)).all)
|
||
|
||
async def create(self, session, user: rbt.User, args) -> LeagueOfLegends:
|
||
name = args.joined()
|
||
|
||
# Connect a new League of Legends account to Royalnet
|
||
log.debug(f"Searching for: {name}")
|
||
summoner = self._riotwatcher.summoner.by_name(region=self.region(), summoner_name=name)
|
||
# Ensure the account isn't already connected to something else
|
||
leagueoflegends = await ru.asyncify(
|
||
session.query(self.alchemy.get(LeagueOfLegends)).filter_by(summoner_id=summoner["id"]).one_or_none)
|
||
if leagueoflegends:
|
||
raise rc.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.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.region(),
|
||
encrypted_summoner_id=summoner["id"])
|
||
# Create database row
|
||
leagueoflegends = self.alchemy.get(LeagueOfLegends)(
|
||
region=self.region(),
|
||
user=user,
|
||
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
|
||
)
|
||
session.add(leagueoflegends)
|
||
return leagueoflegends
|
||
|
||
async def update(self, session, obj: LeagueOfLegends, change: Callable[[str, Any], Awaitable[None]]):
|
||
log.debug(f"Getting summoner data: {obj}")
|
||
summoner = await ru.asyncify(self._riotwatcher.summoner.by_id, region=self.region(),
|
||
encrypted_summoner_id=obj.summoner_id)
|
||
await change("profile_icon_id", summoner["profileIconId"])
|
||
await change("summoner_name", summoner["name"])
|
||
await change("puuid", summoner["puuid"])
|
||
await change("summoner_level", summoner["summonerLevel"])
|
||
await change("summoner_id", summoner["id"])
|
||
await change("account_id", summoner["accountId"])
|
||
log.debug(f"Getting leagues data: {obj}")
|
||
leagues = await ru.asyncify(self._riotwatcher.league.by_summoner, region=self.region(),
|
||
encrypted_summoner_id=obj.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 change("rank_soloq", soloq)
|
||
await change("rank_flexq", flexq)
|
||
await change("rank_twtrq", twtrq)
|
||
await change("rank_tftq", tftq)
|
||
log.debug(f"Getting mastery data: {obj}")
|
||
mastery = await ru.asyncify(self._riotwatcher.champion_mastery.scores_by_summoner,
|
||
region=self.region(),
|
||
encrypted_summoner_id=obj.summoner_id)
|
||
await change("mastery_score", mastery)
|
||
|
||
async def on_increase(self, session, obj: LeagueOfLegends, attribute: str, old: Any, new: Any) -> None:
|
||
if attribute in self.queue_names.keys():
|
||
await self.notify(f"📈 [b]{obj.user}[/b] è salito a {new} su League of Legends ({self.queue_names[attribute]})! Congratulazioni!")
|
||
|
||
async def on_unchanged(self, session, obj: LeagueOfLegends, attribute: str, old: Any, new: Any) -> None:
|
||
pass
|
||
|
||
async def on_decrease(self, session, obj: LeagueOfLegends, attribute: str, old: Any, new: Any) -> None:
|
||
if attribute in self.queue_names.keys():
|
||
await self.notify(f"📉 [b]{obj.user}[/b] è sceso a {new} su League of Legends ({self.queue_names[attribute]}).")
|
||
|
||
async def on_first(self, session, obj: LeagueOfLegends, attribute: str, old: None, new: Any) -> None:
|
||
if attribute in self.queue_names.keys():
|
||
await self.notify(f"🌟 [b]{obj.user}[/b] si è classificato {new} su League of Legends ({self.queue_names[attribute]}!")
|
||
|
||
async def on_reset(self, session, obj: LeagueOfLegends, attribute: str, old: Any, new: None) -> None:
|
||
if attribute in self.queue_names.keys():
|
||
await self.notify(f"⬜️ [b]{obj.user}[/b] non ha più un rank su League of Legends ({self.queue_names[attribute]}).")
|