start progress on lel

royalnet/packs/royal/commands/leagueoflegends.py

@@ -1,6 +1,7 @@
 import typing
 import riotwatcher
 import logging
+import asyncio
 from royalnet.commands import *
 from royalnet.utils import *
 from ..tables import LeagueOfLegends
@@ -23,21 +24,51 @@ class LeagueoflegendsCommand(Command):
     
     _region = "euw1"
 
+    _telegram_group_id = -1001153723135
+
     def __init__(self, interface: CommandInterface):
         super().__init__(interface)
         self._riotwatcher = riotwatcher.RiotWatcher(api_key=self.interface.bot.get_secret("leagueoflegends"))
         ...
 
+    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!
+                # 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
+                # Prepare the message
+                if new_value > old_value:
+                    message = f"📈 [b]{obj.user}[/b] è salito a {new_value} su League of Legends!"
+                else:
+                    message = "📉 [b]{obj.user}[/b] è sceso a {new_value} su League of Legends."
+                # Send the message
+
+    @staticmethod
+    def _change(obj: LeagueOfLegends,
+                attribute_name: str,
+                new_value: typing.Any,
+                callback: typing.Callable[[LeagueOfLegends, str, typing.Any, typing.Any], None]):
+        old_value = obj.__getattribute__(attribute_name)
+        if old_value != new_value:
+            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)
-        lol.profile_icon_id = summoner["profileIconId"]
+        self._change(lol, "profile_icon_id", summoner["profileIconId"], self._notify)
-        lol.summoner_name = summoner["name"]
+        self._change(lol, "summoner_name", summoner["name"], self._notify)
-        lol.puuid = summoner["puuid"]
+        self._change(lol, "puuid", summoner["puuid"], self._notify)
-        lol.summoner_level = summoner["summonerLevel"]
+        self._change(lol, "summoner_level", summoner["summonerLevel"], self._notify)
-        lol.summoner_id = summoner["id"]
+        self._change(lol, "summoner_id", summoner["id"], self._notify)
-        lol.account_id = summoner["accountId"]
+        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
@@ -53,13 +84,21 @@ class LeagueoflegendsCommand(Command):
                 twtrq = LeagueLeague.from_dict(league)
             if league["queueType"] == "RANKED_TFT":
                 tftq = LeagueLeague.from_dict(league)
-        lol.rank_soloq = soloq
+        self._change(lol, "rank_soloq", soloq, self._notify)
-        lol.rank_flexq = flexq
+        self._change(lol, "rank_flexq", flexq, self._notify)
-        lol.rank_twtrq = twtrq
+        self._change(lol, "rank_twtrq", twtrq, self._notify)
-        lol.rank_tftq = tftq
+        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)
-        lol.mastery_score = mastery
+        self._change(lol, "mastery_score", mastery, self._notify)
+
+    async def _updater(self, period: int):
+        while True:
+            session = self.alchemy.Session()
+            lols = session.query(self.alchemy.LeagueOfLegends).all()
+            for lol in lols:
+                await self._update(lol)
+            await asyncio.sleep(period)
 
     def _display(self, lol: LeagueOfLegends):
         string = f"ℹ️ [b]{lol.summoner_name}[/b]\n" \

royalnet/packs/royal/utils/leagueleague.py

@@ -23,7 +23,7 @@ class LeagueLeague:
         self.fresh_blood: bool = fresh_blood
         self.veteran: bool = veteran
 
-    def __str__(self):
+    def __str__(self) -> str:
         emojis = ""
         if self.veteran:
             emojis += "🏆"
@@ -33,11 +33,12 @@ class LeagueLeague:
             emojis += "⭐️"
         return f"[b]{self.tier} {self.rank}[/b] ({self.points} LP) {emojis}"
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"<{self.__class__.__qualname__} {self}>"
 
-    def __eq__(self, other):
+    def __eq__(self, other) -> bool:
-        if isinstance(other, LeagueLeague):
+        if not isinstance(other, LeagueLeague):
+            raise TypeError(f"Can't compare {self.__class__.__qualname__} with {other.__class__.__qualname__}")
         equal = True
         if other.veteran:
             equal &= self.veteran == other.veteran
@@ -58,12 +59,30 @@ class LeagueLeague:
         if other.tier:
             equal &= self.tier == other.tier
         return equal
-        else:
-            raise TypeError(f"Can't compare {self.__class__.__qualname__} with {other.__class__.__qualname__}")
 
-    def __ne__(self, other):
+    def __ne__(self, other) -> bool:
         return not self.__eq__(other)
 
+    def __gt__(self, other) -> bool:
+        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)):
+            raise ValueError("Can't compare partial LeagueLeagues.")
+        if self.tier != other.tier:
+            # Silver is better than Bronze
+            return self.tier > other.tier
+        elif self.rank != other.rank:
+            # Silver I is better than Silver IV
+            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
+        elif self.winrate != other.winrate:
+            # Silver I (100 LP with 60% winrate) is better than Silver I (100 LP with 40% winrate)
+            return self.winrate > other.winrate
+        else:
+            return False
+
     def __bool__(self):
         result = True
         result &= self.veteran is not None