diff --git a/royalpack/commands/__init__.py b/royalpack/commands/__init__.py index 73d1ad3c..79421a77 100644 --- a/royalpack/commands/__init__.py +++ b/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__ diff --git a/royalpack/commands/abstract/linker.py b/royalpack/commands/abstract/linker.py index 20303b82..1c938f75 100644 --- a/royalpack/commands/abstract/linker.py +++ b/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() - log.debug(f"Waiting for: {delay} seconds (delay)") - await aio.sleep(delay) + if index < len(objects) - 1: + delay = self.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) diff --git a/royalpack/commands/brawlhalla.py b/royalpack/commands/brawlhalla.py new file mode 100644 index 00000000..a4d2939b --- /dev/null +++ b/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) diff --git a/royalpack/commands/trionfireali.py b/royalpack/commands/trionfireali.py index f473c29d..2cc98303 100644 --- a/royalpack/commands/trionfireali.py +++ b/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 ..types import DotaRank +from ..halloween2020 import * 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]: - raise NotImplementedError() + async def get_updatables_of_user(self, session, user: rbt.User) -> List[TrionfiStatus]: + return [user.halloween2020] if user.halloween2020 else [] - async def get_updatables(self, session) -> List[Dota]: - raise NotImplementedError() + async def get_updatables(self, session) -> List[TrionfiStatus]: + 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]: - raise rc.InvalidInputError("Trionfi Reali accounts are automatically linked from Steam.") + data: Optional[rc.CommandData] = None) -> Optional[TrionfiStatus]: + raise rc.InvalidInputError("N⊕n è qui che inizia il mister⊕.") - async def update(self, session, obj: Steam, change: Callable[[str, Any], Awaitable[None]]): - raise NotImplementedError() + async def update(self, session, obj: TrionfiStatus, change: Callable[[str, Any], Awaitable[None]]): + 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 diff --git a/royalpack/halloween2020/__init__.py b/royalpack/halloween2020/__init__.py index e69de29b..7a6cc67b 100644 --- a/royalpack/halloween2020/__init__.py +++ b/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" +] diff --git a/royalpack/halloween2020/check.py b/royalpack/halloween2020/check.py index 22742b5a..90cd877f 100644 --- a/royalpack/halloween2020/check.py +++ b/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_played_free_games": True, - "include_free_sub": True, + "include_appinfo": "true", + "include_played_free_games": "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: - return (await self.first.check(status)) or (await self.second.check(status)) + def __repr__(self): + 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: - return (await self.first.check(status)) and (await self.second.check(status)) + def __repr__(self): + 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)) diff --git a/royalpack/halloween2020/trionfilist.py b/royalpack/halloween2020/trionfilist.py index 987ce39e..98ad56b6 100644 --- a/royalpack/halloween2020/trionfilist.py +++ b/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(), ), -) \ No newline at end of file +] \ No newline at end of file diff --git a/royalpack/halloween2020/trionfoinfo.py b/royalpack/halloween2020/trionfoinfo.py index 61c63cd0..ea5d8ff5 100644 --- a/royalpack/halloween2020/trionfoinfo.py +++ b/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 { diff --git a/royalpack/tables/__init__.py b/royalpack/tables/__init__.py index 26eb3e0d..1562c32a 100644 --- a/royalpack/tables/__init__.py +++ b/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__