Add Brawlhalla to the statsupdater

db.py

@@ -1,6 +1,8 @@
 import datetime
 import logging
 import os
+import time
+
 import coloredlogs
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker, relationship
@@ -1269,6 +1271,67 @@ class BindingOfIsaacRun(Base):
         return f"<db.BindingOfIsaacRun {self.player_id}: {self.score}>"
 
 
+class Brawlhalla(Base, Mini):
+    __tablename__ = "brawlhalla"
+
+    steam_id = Column(String, ForeignKey("steam.steam_id"), primary_key=True)
+    steam = relationship("Steam", backref="brawlhalla", lazy="joined")
+    brawlhalla_id = Column(BigInteger)
+    name = Column(String)
+
+    level = Column(Integer)
+
+    main_legend_name = Column(String)
+    main_legend_level = Column(Integer)
+
+    ranked_plays = Column(Integer)
+    ranked_wins = Column(Integer)
+    rating = Column(Integer)
+
+    best_team_name = Column(String)
+    best_team_rating = Column(Integer)
+
+    def __repr__(self):
+        return f"<db.Brawlhalla {self.name}>"
+
+    @staticmethod
+    def init_table():
+        session = Session()
+        steam = session.query(Steam).all()
+        for user in steam:
+            j = requests.get("https://api.brawlhalla.com/search", params={
+                "steamid": user.steam_id,
+                "api_key": config["Brawlhalla"]["brawlhalla_api_key"]
+            }).json()
+            if not j:
+                continue
+            b = session.query(Brawlhalla).filter_by(steam=user).one_or_none()
+            if b is None:
+                b = Brawlhalla(steam_id=user.steam_id, brawlhalla_id=j["brawlhalla_id"], name=j["name"])
+            session.add(b)
+            time.sleep(1)
+        session.commit()
+
+    def update(self):
+        j = requests.get(f"https://api.brawlhalla.com/player/{self.brawlhalla_id}/stats?api_key={config['Brawlhalla']['brawlhalla_api_key']}").json()
+        self.name = j["name"]
+        self.level = j["level"]
+        main_legend = max(j["legends"], key=lambda l: l["level"])
+        self.main_legend_level = main_legend["level"]
+        self.main_legend_name = main_legend["legend_name_key"]
+        j = requests.get(f"https://api.brawlhalla.com/player/{self.brawlhalla_id}/ranked?api_key={config['Brawlhalla']['brawlhalla_api_key']}").json()
+        self.ranked_plays = j["games"]
+        self.ranked_wins = j["wins"]
+        rating = DirtyDelta(self.rating)
+        rating.value = j["rating"]
+        best_team_data = Dirty((self.best_team_name, self.best_team_rating))
+        current_best_team = max(j["2v2"], key=lambda t: t["rating"])
+        self.best_team_name = current_best_team["name"]
+        self.best_team_rating = current_best_team["rating"]
+        best_team_data.value = (self.best_team_name, self.best_team_rating)
+        return rating, best_team_data
+
+
 # If run as script, create all the tables in the db
 if __name__ == "__main__":
     print("Creating new tables...")

statsupdate.py

@@ -11,7 +11,8 @@ import telegram
 import sys
 import coloredlogs
 import requests
-from utils import Dirty, DirtyDelta
+import strings
+from utils import Dirty, DirtyDelta, reply_msg
 
 logging.getLogger().disabled = True
 logger = logging.getLogger(__name__)
@@ -29,6 +30,7 @@ sentry = raven.Client(config["Sentry"]["token"],
                       hook_libraries=[])
 
 telegram_bot = telegram.Bot(config["Telegram"]["bot_token"])
+main_chat_id = config["Telegram"]["main_group"]
 
 
 def update_block(session: db.Session, block: list, delay: float = 0, change_callback: typing.Callable = None):
@@ -126,9 +128,30 @@ def osu_pp_change(item, change: typing.Tuple[DirtyDelta, DirtyDelta, DirtyDelta,
         sentry.captureException()
 
 
+def brawlhalla_rank_change(item, change: typing.Tuple[DirtyDelta, Dirty]):
+    solo, team = change
+    try:
+        if solo.delta >= 10:
+            reply_msg(telegram_bot, main_chat_id, strings.STATSUPDATE.BRAWLHALLA.SOLO,
+                      username=item.steam.royal.username,
+                      rating=solo.value,
+                      delta=solo.delta)
+        if team.is_dirty():
+            reply_msg(telegram_bot, main_chat_id, strings.STATSUPDATE.BRAWLHALLA.TEAM,
+                      username=item.steam.royal.username,
+                      rating=team[1],
+                      teamname=team[0])
+    except Exception:
+        logger.warning(f"Couldn't notify on Telegram: {item}")
+        sentry.captureException()
+
+
 def process():
     while True:
         session = db.Session()
+        logger.info("Now updating League of Legends data.")
+        update_block(session, session.query(db.Brawlhalla).all(), delay=5, change_callback=brawlhalla_rank_change)
+        session.commit()
         logger.info("Now updating osu! data.")
         update_block(session, session.query(db.Osu).all(), delay=5, change_callback=osu_pp_change)
         session.commit()

strings.py

@@ -248,6 +248,13 @@ class SPELL:
         INVALID_SYNTAX = "⚠ Non hai specificato la magia di cui vuoi conoscere i dettagli!\nSintassi: <code>/spell (nome)</code>"
 
 
+# Game stats updates
+class STATSUPDATE:
+    class BRAWLHALLA:
+        SOLO = "✳️ {username} ha ora <b>{rating}</b> ({delta}) Elo 1v1 su Brawlhalla!"
+        TEAM = "✳️ {username} ha ora <b>{rating}</b> Elo 2v2 con il team <i>{teamname}</i> di Brawlhalla!"
+
+
 # Secondo me, è colpa delle stringhe.
 SMECDS = "🤔 Secondo me, è colpa {ds}."
 

template_config.ini

@@ -67,3 +67,7 @@ ppy_api_key =
 [Sentry]
 # Sentry error-reporting token obtainable at https://sentry.io
 token =
+
+[Brawlhalla]
+# Brawlhalla API key, requires an email, obtainable at http://dev.brawlhalla.com
+brawlhalla_api_key =