ryg/royalnet
ryg/royalnet
1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-23 19:44:20 +00:00
Code Activity

Complete updater

Browse source
This commit is contained in:
Steffo 2019-10-29 21:52:31 +01:00
parent d49f1f39eb
commit 3a7e4d72f9
4 changed files with 91 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

112
royalnet/packs/royal/commands/leagueoflegends.py
View file

@ -2,12 +2,12 @@ import typing
import riotwatcher
import logging
import asyncio
import sentry_sdk
from royalnet.commands import *
from royalnet.utils import *
from ..tables import LeagueOfLegends
from ..utils import LeagueLeague
log = logging.getLogger(__name__)
@ -19,9 +19,9 @@ class LeagueoflegendsCommand(Command):
description: str = "Connetti un account di League of Legends a un account Royalnet, e visualizzane le statistiche."
syntax = "[nomeevocatore]"
tables = {LeagueOfLegends}
_region = "euw1"
_telegram_group_id = -1001153723135
@ -29,13 +29,14 @@ class LeagueoflegendsCommand(Command):
def __init__(self, interface: CommandInterface):
super().__init__(interface)
self._riotwatcher = riotwatcher.RiotWatcher(api_key=self.interface.bot.get_secret("leagueoflegends"))
...
if self.interface.name == "telegram":
self.loop.create_task(self._updater(900))
def _notify(self,
obj: LeagueOfLegends,
attribute_name: str,
old_value: typing.Any,
new_value: typing.Any):
async def _notify(self,
obj: LeagueOfLegends,
attribute_name: str,
old_value: typing.Any,
new_value: typing.Any):
if self.interface.name == "telegram":
if isinstance(old_value, LeagueLeague):
# This is a rank change!
@ -44,37 +45,46 @@ class LeagueoflegendsCommand(Command):
return
# Prepare the message
if new_value > old_value:
message = f"📈 [b]{obj.user}[/b] è salito a {new_value} su League of Legends!"
message = f"📈 [b]{obj.user}[/b] è salito a {new_value} su League of Legends! Congratulazioni!"
else:
message = "📉 [b]{obj.user}[/b] è sceso a {new_value} su League of Legends."
message = f"📉 [b]{obj.user}[/b] è sceso a {new_value} su League of Legends."
# Send the message
client = self.interface.bot.client
await self.interface.bot.safe_api_call(client.send_message,
chat_id=self._telegram_group_id,
text=telegram_escape(message),
parse_mode="HTML",
disable_webpage_preview=True)
@staticmethod
def _change(obj: LeagueOfLegends,
attribute_name: str,
new_value: typing.Any,
callback: typing.Callable[[LeagueOfLegends, str, typing.Any, typing.Any], None]):
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:
callback(obj, attribute_name, 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 = self._riotwatcher.summoner.by_id(region=self._region, encrypted_summoner_id=lol.summoner_id)
self._change(lol, "profile_icon_id", summoner["profileIconId"], self._notify)
self._change(lol, "summoner_name", summoner["name"], self._notify)
self._change(lol, "puuid", summoner["puuid"], self._notify)
self._change(lol, "summoner_level", summoner["summonerLevel"], self._notify)
self._change(lol, "summoner_id", summoner["id"], self._notify)
self._change(lol, "account_id", summoner["accountId"], self._notify)
summoner = await asyncify(self._riotwatcher.summoner.by_id, region=self._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 = self._riotwatcher.league.by_summoner(region=self._region, encrypted_summoner_id=lol.summoner_id)
soloq = None
flexq = None
twtrq = None
tftq = None
leagues = await asyncify(self._riotwatcher.league.by_summoner, region=self._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)
@ -84,20 +94,32 @@ class LeagueoflegendsCommand(Command):
twtrq = LeagueLeague.from_dict(league)
if league["queueType"] == "RANKED_TFT":
tftq = LeagueLeague.from_dict(league)
self._change(lol, "rank_soloq", soloq, self._notify)
self._change(lol, "rank_flexq", flexq, self._notify)
self._change(lol, "rank_twtrq", twtrq, self._notify)
self._change(lol, "rank_tftq", tftq, self._notify)
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 = self._riotwatcher.champion_mastery.scores_by_summoner(region=self._region, encrypted_summoner_id=lol.summoner_id)
self._change(lol, "mastery_score", mastery, self._notify)
mastery = await asyncify(self._riotwatcher.champion_mastery.scores_by_summoner, region=self._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.LeagueOfLegends).all()
for lol in lols:
await self._update(lol)
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)
def _display(self, lol: LeagueOfLegends):
@ -132,10 +154,10 @@ class LeagueoflegendsCommand(Command):
# 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 = None
flexq = None
twtrq = None
tftq = None
soloq = LeagueLeague()
flexq = LeagueLeague()
twtrq = LeagueLeague()
tftq = LeagueLeague()
for league in leagues:
if league["queueType"] == "RANKED_SOLO_5x5":
soloq = LeagueLeague.from_dict(league)
@ -173,10 +195,14 @@ class LeagueoflegendsCommand(Command):
# Update and display the League of Legends stats for the current account
if len(author.leagueoflegends) == 0:
raise CommandError("Nessun account di League of Legends trovato.")
for account in author.leagueoflegends:
await self._update(account)
await data.session_commit()
message = ""
for account in author.leagueoflegends:
message += self._display(account)
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)

12
royalnet/packs/royal/utils/leagueleague.py
View file

@ -31,12 +31,14 @@ class LeagueLeague:
emojis += "🔥"
if self.fresh_blood:
emojis += "⭐️"
return f"[b]{self.tier} {self.rank}[/b] ({self.points} LP) {emojis}"
return f"[b]{self.tier} {self.rank}[/b] ({self.points} LP){' ' if emojis else ''}{emojis}"
def __repr__(self) -> str:
return f"<{self.__class__.__qualname__} {self}>"
def __eq__(self, other) -> bool:
if other is None:
return False
if not isinstance(other, LeagueLeague):
raise TypeError(f"Can't compare {self.__class__.__qualname__} with {other.__class__.__qualname__}")
equal = True
@ -64,6 +66,8 @@ class LeagueLeague:
return not self.__eq__(other)
def __gt__(self, other) -> bool:
if other is None:
return True
if not isinstance(other, LeagueLeague):
raise TypeError(f"Can't compare {self.__class__.__qualname__} with {other.__class__.__qualname__}")
if not (bool(self) and bool(other)):
@ -73,7 +77,7 @@ class LeagueLeague:
return self.tier > other.tier
elif self.rank != other.rank:
# Silver I is better than Silver IV
return self.rank < other.rank
return self.rank > other.rank
elif self.points != other.points:
# Silver I (100 LP) is better than Silver I (0 LP)
return self.points > other.points
@ -118,8 +122,8 @@ class LeagueLeague:
@classmethod
def from_dict(cls, d: dict):
return cls(
tier=d["tier"],
rank=d["rank"],
tier=LeagueTier.from_string(d["tier"]),
rank=LeagueRank.from_string(d["rank"]),
points=d["leaguePoints"],
wins=d["wins"],
losses=d["losses"],

7
royalnet/packs/royal/utils/leaguerank.py
View file

@ -12,3 +12,10 @@ class LeagueRank(enum.Enum):
def __repr__(self):
return f"{self.__class__.__qualname__}.{self.name}"
def __gt__(self, other):
return self.value < other.value
@classmethod
def from_string(cls, string: str):
return cls.__members__.get(string)

7
royalnet/packs/royal/utils/leaguetier.py
View file

@ -17,3 +17,10 @@ class LeagueTier(enum.Enum):
def __repr__(self):
return f"{self.__class__.__qualname__}.{self.name}"
def __gt__(self, other):
return self.value > other.value
@classmethod
def from_string(cls, string: str):
return cls.__members__.get(string)