Complete trionfi backend

royalpack/commands/__init__.py

@@ -37,6 +37,7 @@ from .steampowered import SteampoweredCommand
 from .treasure import TreasureCommand
 from .trivia import TriviaCommand
 from .osu import OsuCommand
+from .trionfireali import TrionfirealiCommand
 
 # Enter the commands of your Pack here!
 available_commands = [
@@ -78,6 +79,7 @@ available_commands = [
     TreasureCommand,
     TriviaCommand,
     OsuCommand,
+    TrionfirealiCommand,
 ]
 
 # Don't change this, it should automatically generate __all__

royalpack/commands/abstract/linker.py

@@ -165,7 +165,7 @@ class LinkerCommand(rc.Command, metaclass=abc.ABCMeta):
             session = self.alchemy.Session()
             objects = await self.get_updatables(session)
 
-            for obj in objects:
+            for index, obj in enumerate(objects):
                 log.debug(f"Updating: {obj} ({self.name})")
 
                 async def change(attribute: str, value: Any):
@@ -182,9 +182,10 @@ class LinkerCommand(rc.Command, metaclass=abc.ABCMeta):
                 except Exception as e:
                     ru.sentry_exc(e)
 
-                delay = self.delay()
+                if index < len(objects) - 1:
-                log.debug(f"Waiting for: {delay} seconds (delay)")
+                    delay = self.delay()
-                await aio.sleep(delay)
+                    log.debug(f"Waiting for: {delay} seconds (delay)")
+                    await aio.sleep(delay)
 
             log.debug(f"Committing updates: {self.name}")
             await ru.asyncify(session.commit)

royalpack/commands/brawlhalla.py

@@ -0,0 +1,169 @@
+from typing import *
+
+import asyncio
+import logging
+import aiohttp
+
+from royalnet.backpack import tables as rbt
+import royalnet.commands as rc
+import royalnet.utils as ru
+from sqlalchemy import or_, and_
+
+from .abstract.linker import LinkerCommand
+from ..tables import Steam, Brawlhalla, BrawlhallaDuo
+from ..types import BrawlhallaRank, BrawlhallaMetal, BrawlhallaTier, Updatable
+
+log = logging.getLogger(__name__)
+
+
+class BrawlhallaCommand(LinkerCommand):
+    name: str = "brawlhalla"
+
+    aliases = ["bh", "bruhalla", "bruhlalla"]
+
+    description: str = "Visualizza le tue statistiche di Brawlhalla."
+
+    syntax: str = ""
+
+    def token(self):
+        return self.config['brawlhalla']['token']
+
+    async def get_updatables_of_user(self, session, user: rbt.User) -> List[Brawlhalla]:
+        return user.steam
+
+    async def get_updatables(self, session) -> List[Brawlhalla]:
+        return await ru.asyncify(session.query(self.alchemy.get(Steam)).all)
+
+    async def create(self,
+                     session,
+                     user: rbt.User,
+                     args: rc.CommandArgs,
+                     data: Optional[rc.CommandData] = None) -> Optional[Brawlhalla]:
+        raise rc.InvalidInputError("Brawlhalla accounts are automatically linked from Steam.")
+
+    async def update(self, session, obj, change: Callable[[str, Any], Awaitable[None]]):
+        BrawlhallaT = self.alchemy.get(Brawlhalla)
+        DuoT = self.alchemy.get(BrawlhallaDuo)
+        log.info(f"Updating: {obj}")
+        async with aiohttp.ClientSession() as hcs:
+            bh: Brawlhalla = obj.brawlhalla
+            if bh is None:
+                log.debug(f"Checking if player has an account...")
+                async with hcs.get(f"https://api.brawlhalla.com/search?steamid={obj.steamid.as_64}&api_key={self.token()}") as response:
+                    if response.status != 200:
+                        raise rc.ExternalError(f"Brawlhalla API /search returned {response.status}!")
+                    j = await response.json()
+                    if j == {} or j == []:
+                        log.debug("No account found.")
+                        return
+                    bh = BrawlhallaT(
+                        steam=obj,
+                        brawlhalla_id=j["brawlhalla_id"],
+                        name=j["name"]
+                    )
+                    session.add(bh)
+                    session.flush()
+
+            async with hcs.get(f"https://api.brawlhalla.com/player/{bh.brawlhalla_id}/ranked?api_key={self.token()}") as response:
+                if response.status != 200:
+                    raise rc.ExternalError(f"Brawlhalla API /ranked returned {response.status}!")
+                j = await response.json()
+                if j == {} or j == []:
+                    log.debug("No ranked info found.")
+                else:
+                    await self._change(session=session, obj=bh, attribute="rating_1v1", new=j["rating"])
+                    metal_name, tier_name = j["tier"].split(" ", 1)
+                    metal = BrawlhallaMetal[metal_name.upper()]
+                    tier = BrawlhallaTier(int(tier_name))
+                    rank = BrawlhallaRank(metal=metal, tier=tier)
+                    await self._change(session=session, obj=bh, attribute="rank_1v1", new=rank)
+
+                    for jduo in j.get("2v2", []):
+                        bhduo: Optional[BrawlhallaDuo] = await ru.asyncify(
+                            session.query(DuoT)
+                                .filter(
+                                    or_(
+                                        and_(
+                                            DuoT.id_one == jduo["brawlhalla_id_one"],
+                                            DuoT.id_two == jduo["brawlhalla_id_two"]
+                                        ),
+                                        and_(
+                                            DuoT.id_one == jduo["brawlhalla_id_two"],
+                                            DuoT.id_two == jduo["brawlhalla_id_one"]
+                                        )
+                                    )
+                                )
+                                .one_or_none
+                        )
+                        if bhduo is None:
+                            if bh.brawlhalla_id == jduo["brawlhalla_id_one"]:
+                                otherbh: Optional[Brawlhalla] = await ru.asyncify(
+                                    session.query(BrawlhallaT).get, jduo["brawlhalla_id_two"]
+                                )
+                            else:
+                                otherbh: Optional[Brawlhalla] = await ru.asyncify(
+                                    session.query(BrawlhallaT).get, jduo["brawlhalla_id_one"]
+                                )
+                            if otherbh is None:
+                                continue
+                            bhduo = DuoT(
+                                one=bh,
+                                two=otherbh,
+                            )
+
+                            session.add(bhduo)
+                        await self._change(session=session, obj=bhduo, attribute="rating_2v2", new=jduo["rating"])
+                        metal_name, tier_name = jduo["tier"].split(" ", 1)
+                        metal = BrawlhallaMetal[metal_name.upper()]
+                        tier = BrawlhallaTier(int(tier_name))
+                        rank = BrawlhallaRank(metal=metal, tier=tier)
+                        await self._change(session=session, obj=bhduo, attribute="rank_2v2", new=rank)
+
+    async def on_increase(self, session, obj: Union[Brawlhalla, BrawlhallaDuo], attribute: str, old: Any, new: Any) -> None:
+        if attribute == "rank_1v1":
+            await self.notify(f"📈 [b]{obj.steam.user}[/b] è salito a [b]{new}[/b] ({obj.rating_1v1} MMR) in 1v1 su Brawlhalla! Congratulazioni!")
+        elif attribute == "rank_2v2":
+            await self.notify(f"📈 [b]{obj.one.steam.user}[/b] e [b]{obj.two.steam.user}[/b] sono saliti a [b]{new}[/b] ({obj.rating_2v2} MMR) in 2v2 su Brawlhalla! Congratulazioni!")
+
+    async def on_unchanged(self, session, obj: Union[Brawlhalla, BrawlhallaDuo], attribute: str, old: Any, new: Any) -> None:
+        pass
+
+    async def on_decrease(self, session, obj: Union[Brawlhalla, BrawlhallaDuo], attribute: str, old: Any, new: Any) -> None:
+        if attribute == "rank_1v1":
+            await self.notify(f"📉 [b]{obj.steam.user}[/b] è sceso a [b]{new}[/b] ({obj.rating_1v1} MMR) in 1v1 su Brawlhalla.")
+        elif attribute == "rank_2v2":
+            await self.notify(f"📉 [b]{obj.one.steam.user}[/b] e [b]{obj.two.steam.user}[/b] sono scesi a [b]{new}[/b] ({obj.rating_2v2} MMR) in 2v2 su Brawlhalla.")
+
+    async def on_first(self, session, obj: Union[Brawlhalla, BrawlhallaDuo], attribute: str, old: None, new: Any) -> None:
+        if attribute == "rank_1v1":
+            await self.notify(f"🌟 [b]{obj.steam.user}[/b] si è classificato a [b]{new}[/b] ({obj.rating_1v1} MMR) in 1v1 su Brawlhalla!")
+        elif attribute == "rank_2v2":
+            await self.notify(f"🌟 [b]{obj.one.steam.user}[/b] e [b]{obj.two.steam.user}[/b] si sono classificati a [b]{new}[/b] ({obj.rating_2v2} MMR) in 2v2 su Brawlhalla!")
+
+    async def on_reset(self, session, obj: Union[Brawlhalla, BrawlhallaDuo], attribute: str, old: Any, new: None) -> None:
+        if attribute == "rank_1v1":
+            await self.notify(f"⬜️ [b]{obj.steam.user}[/b] non ha più un rank su Brawlhalla.")
+        elif attribute == "rank_2v2":
+            await self.notify(f"⬜️ [b]{obj.one.steam.user}[/b] e [b]{obj.two.steam.user}[/b] non hanno più un rank su Brawlhalla.")
+
+    def describe(self, obj: Steam) -> str:
+        bh = obj.brawlhalla
+
+        string = [f"ℹ️ [b]{bh.name}[/b]", ""]
+
+        if bh.rank_1v1:
+            string.append("👤 [b]1v1[/b]")
+            string.append(f"[b]{bh.rank_1v1}[/b] ({bh.rating_1v1} MMR)")
+            string.append("")
+
+        if len(bh.duos) != 0:
+            string.append(f"👥 [b]2v2[/b]")
+
+        for duo in sorted(bh.duos, key=lambda d: -d.rating_2v2):
+            other = duo.other(bh)
+            string.append(f"Con [b]{other.steam.user}[/b]: [b]{duo.rank_2v2}[/b] ({duo.rating_2v2} MMR)")
+
+        if len(bh.duos) != 0:
+            string.append("")
+
+        return "\n".join(string)

royalpack/commands/trionfireali.py

@@ -1,20 +1,17 @@
 from typing import *
 import logging
-import steam.webapi
 import royalnet.commands as rc
 import royalnet.utils as ru
-import requests
 from royalnet.backpack import tables as rbt
 from .abstract.linker import LinkerCommand
+import asyncio
+import datetime
 
-from ..tables import Steam, Dota
+from ..halloween2020 import *
-from ..types import DotaRank
 
 log = logging.getLogger(__name__)
 
 
-
-
 class TrionfirealiCommand(LinkerCommand):
     name: str = "trionfireali"
 
@@ -22,36 +19,41 @@ class TrionfirealiCommand(LinkerCommand):
 
     syntax: str = ""
 
-    def describe(self, obj: Steam) -> str:
+    def describe(self, obj: TrionfiStatus) -> str:
         raise NotImplementedError()
 
-    async def get_updatables_of_user(self, session, user: rbt.User) -> List[Dota]:
+    async def get_updatables_of_user(self, session, user: rbt.User) -> List[TrionfiStatus]:
-        raise NotImplementedError()
+        return [user.halloween2020] if user.halloween2020 else []
 
-    async def get_updatables(self, session) -> List[Dota]:
+    async def get_updatables(self, session) -> List[TrionfiStatus]:
-        raise NotImplementedError()
+        return await ru.asyncify(session.query(self.alchemy.get(TrionfiStatus)).all)
 
     async def create(self,
                      session,
                      user: rbt.User,
                      args: rc.CommandArgs,
-                     data: Optional[rc.CommandData] = None) -> Optional[Dota]:
+                     data: Optional[rc.CommandData] = None) -> Optional[TrionfiStatus]:
-        raise rc.InvalidInputError("Trionfi Reali accounts are automatically linked from Steam.")
+        raise rc.InvalidInputError("N⊕n è qui che inizia il mister⊕.")
 
-    async def update(self, session, obj: Steam, change: Callable[[str, Any], Awaitable[None]]):
+    async def update(self, session, obj: TrionfiStatus, change: Callable[[str, Any], Awaitable[None]]):
-        raise NotImplementedError()
+        for trionfo in trionfilist:
+            check_result = await trionfo.check.check(obj, key=self.config["steampowered"]["token"])
+            log.debug(f"{trionfo.check}: {check_result}")
+            if check_result is True:
+                obj.__setattr__(trionfo.variable, datetime.datetime.now())
+            await asyncio.sleep(1)
 
-    async def on_increase(self, session, obj: Dota, attribute: str, old: Any, new: Any) -> None:
+    async def on_increase(self, session, obj: TrionfiStatus, attribute: str, old: Any, new: Any) -> None:
         pass
 
-    async def on_unchanged(self, session, obj: Dota, attribute: str, old: Any, new: Any) -> None:
+    async def on_unchanged(self, session, obj: TrionfiStatus, attribute: str, old: Any, new: Any) -> None:
         pass
 
-    async def on_decrease(self, session, obj: Dota, attribute: str, old: Any, new: Any) -> None:
+    async def on_decrease(self, session, obj: TrionfiStatus, attribute: str, old: Any, new: Any) -> None:
         pass
 
-    async def on_first(self, session, obj: Dota, attribute: str, old: None, new: Any) -> None:
+    async def on_first(self, session, obj: TrionfiStatus, attribute: str, old: None, new: Any) -> None:
         pass
 
-    async def on_reset(self, session, obj: Dota, attribute: str, old: Any, new: None) -> None:
+    async def on_reset(self, session, obj: TrionfiStatus, attribute: str, old: Any, new: None) -> None:
         pass

royalpack/halloween2020/__init__.py

@@ -0,0 +1,13 @@
+from .check import *
+from .trionfilist import *
+from .trionfoinfo import *
+from .trionfistatus import *
+
+__all__ = [
+    "Check",
+    "CheckPlayedSteamGame",
+    "CheckAchievementSteamGame",
+    "TrionfiStatus",
+    "TrionfoInfo",
+    "trionfilist"
+]

royalpack/halloween2020/check.py

@@ -1,13 +1,19 @@
 from typing import *
 import abc
 import aiohttp
+import logging
+import royalnet.commands as rc
 
 if TYPE_CHECKING:
     from .trionfistatus import TrionfiStatus
 
 
+log = logging.getLogger(__name__)
+
+
 __all__ = [
     "Check",
+    "NullCheck",
     "CheckPlayedSteamGame",
     "CheckAchievementSteamGame",
 ]
@@ -15,7 +21,7 @@ __all__ = [
 
 class Check(metaclass=abc.ABCMeta):
     @abc.abstractmethod
-    async def check(self, status: "TrionfiStatus") -> bool:
+    async def check(self, status: "TrionfiStatus", key: str) -> bool:
         raise NotImplementedError()
 
     def __or__(self, other: "Check"):
@@ -25,25 +31,39 @@ class Check(metaclass=abc.ABCMeta):
         return CheckAnd(self, other)
 
 
+class NullCheck(Check):
+    def __repr__(self):
+        return f"{self.__class__.__name__}()"
+
+    async def check(self, status: "TrionfiStatus", key: str) -> bool:
+        return False
+
+
 class CheckPlayedSteamGame(Check):
     def __init__(self, appid: int, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.appid: int = appid
 
-    async def check(self, status: "TrionfiStatus") -> bool:
+    def __repr__(self):
+        return f"{self.__class__.__name__}({self.appid=})"
+
+    async def check(self, status: "TrionfiStatus", key: str) -> bool:
+        log.debug(f"{self}")
         async with aiohttp.ClientSession() as ah_session:
             # noinspection PyProtectedMember
             async with ah_session.get("https://api.steampowered.com/IPlayerService/GetOwnedGames/v1/",
                                       params={
                                           "steamid": status._steamid,
-                                          "include_appinfo": True,
+                                          "include_appinfo": "true",
-                                          "include_played_free_games": True,
+                                          "include_played_free_games": "true",
-                                          "include_free_sub": True,
+                                          "include_free_sub": "true",
                                           "appids_filter": self.appid,
+                                          "key": key,
                                       }) as response:
                 try:
                     j = await response.json()
-                except Exception:
+                except Exception as e:
+                    log.error(f"{e}")
                     return False
 
                 games = j["response"]["games"]
@@ -61,19 +81,26 @@ class CheckAchievementSteamGame(Check):
         self.appid: int = appid
         self.achivement_name: str = achievement_name
 
-    async def check(self, status: "TrionfiStatus") -> bool:
+    def __repr__(self):
+        return f"{self.__class__.__name__}({self.appid=}, {self.achivement_name=})"
+
+    async def check(self, status: "TrionfiStatus", key: str) -> bool:
+        log.debug(f"{self}")
         async with aiohttp.ClientSession() as ah_session:
             # noinspection PyProtectedMember
             async with ah_session.get("http://api.steampowered.com/ISteamUserStats/GetPlayerAchievements/v1/",
                                       params={
                                           "steamid": status._steamid,
                                           "appid": self.appid,
+                                          "key": key,
                                       }) as response:
                 try:
                     j = await response.json()
-                except Exception:
+                except Exception as e:
+                    log.error(f"{e}")
                     return False
                 if not j["playerstats"]["success"]:
+                    log.warning(f"{j}")
                     return False
 
                 achievements = j["playerstats"]["achievements"]
@@ -90,8 +117,12 @@ class CheckOr(Check):
         self.first: Check = first
         self.second: Check = second
 
-    async def check(self, status: "TrionfiStatus") -> bool:
+    def __repr__(self):
-        return (await self.first.check(status)) or (await self.second.check(status))
+        return f"{self.first} or {self.second}"
+
+    async def check(self, status: "TrionfiStatus", key: str) -> bool:
+        log.debug(f"{self}")
+        return (await self.first.check(status, key)) or (await self.second.check(status, key))
 
 
 class CheckAnd(Check):
@@ -100,5 +131,9 @@ class CheckAnd(Check):
         self.first: Check = first
         self.second: Check = second
 
-    async def check(self, status: "TrionfiStatus") -> bool:
+    def __repr__(self):
-        return (await self.first.check(status)) and (await self.second.check(status))
+        return f"{self.first} and {self.second}"
+
+    async def check(self, status: "TrionfiStatus", key: str) -> bool:
+        log.debug(f"{self}")
+        return (await self.first.check(status, key)) and (await self.second.check(status, key))

royalpack/halloween2020/trionfilist.py

@@ -1,7 +1,8 @@
+from typing import *
 from .trionfoinfo import TrionfoInfo
 from .check import *
 
-trionfilist = (
+trionfilist: List[TrionfoInfo] = [
     TrionfoInfo(
         variable="zero",
         title="o",
@@ -9,7 +10,7 @@ trionfilist = (
         name="Il Folle",
         puzzle="UN VIAGGIO TI ATTENDE",
         objective="Partecipa ai Trionfi Reali.",
-        check=None,
+        check=NullCheck(),
     ),
     TrionfoInfo(
         variable="i",
@@ -18,7 +19,7 @@ trionfilist = (
         name="Il Mago",
         puzzle="L'ULTIMO GIORNO",
         objective="Trova una /spell che possa fare almeno 250 danni.",
-        check=None,
+        check=NullCheck(),
     ),
     TrionfoInfo(
         variable="ii",
@@ -75,7 +76,7 @@ trionfilist = (
         puzzle="SOPRA UN CARRO",
         objective="Gioca 5 incontri a [url=https://store.steampowered.com/app/326460/ShellShock_Live]ShellShock Live["
                   "/url].",
-        check=CheckPlayedSteamGame(326460, "play5")
+        check=CheckAchievementSteamGame(326460, "play5")
     ),
     TrionfoInfo(
         variable="viii",
@@ -103,7 +104,7 @@ trionfilist = (
         name="La Fortuna",
         puzzle="LA CASA DEI GIOCHI",
         objective="Chiedi a Royal Bot di predire il tuo futuro.",
-        check=None,
+        check=NullCheck(),
     ),
     TrionfoInfo(
         variable="xi",
@@ -209,6 +210,6 @@ trionfilist = (
         name="Il Mondo",
         puzzle="""44°35'45.0"N 11°02'58.9"E""",
         objective="Vinci la partita a Trionfi Reali.",
-        check=None,
+        check=NullCheck(),
     ),
-)
+]

royalpack/halloween2020/trionfoinfo.py

@@ -13,14 +13,14 @@ class TrionfoInfo:
                  name: str,
                  objective: str,
                  puzzle: str,
-                 check: Optional["Check"]):
+                 check: "Check"):
         self.variable: str = variable
         self.title: str = title
         self.roman: str = roman
         self.name: str = name
         self.objective: str = objective
         self.puzzle: str = puzzle
-        self.check: Optional["Check"] = check
+        self.check: "Check" = check
 
     def json_anonymous(self) -> ru.JSON:
         return {

royalpack/tables/__init__.py

@@ -19,7 +19,7 @@ from .mmresponse import MMResponse
 from .cvstats import Cvstats
 from .treasure import Treasure
 from .osu import Osu
-from ..halloween2020.trionfistatus import Halloween2020
+from ..halloween2020.trionfistatus import TrionfiStatus
 
 # Enter the tables of your Pack here!
 available_tables = [
@@ -43,7 +43,7 @@ available_tables = [
     Cvstats,
     Treasure,
     Osu,
-    Halloween2020,
+    TrionfiStatus,
 ]
 
 # Don't change this, it should automatically generate __all__