extract

src/extract.py

@@ -0,0 +1,105 @@
+import psycopg2 as pg
+from os import environ as env
+from typing import TypeVar
+
+T = TypeVar('T')
+U = TypeVar('U')
+
+
+def connect():
+    return pg.connect(dbname=env.get("POSTGRES_DB"),
+                      user=env.get("POSTGRES_USER"),
+                      password=env.get("POSTGRES_PASSWORD"),
+                      host=env.get("POSTGRES_HOST"),
+                      port=env.get("POSTGRES_PORT"))
+
+def extract_cols(data: dict, cols: list[str]) -> dict:
+    return {key: data.get(key) for key in cols}
+
+def score_to_text(score: list[int]) -> str:
+    return "-".join(map(str,score))
+
+# maybe could get columns form db
+def cols_from_player_stats(player_stats: dict) -> dict:
+    cols =  ["id", "gameId", "gameFinishedAt", "steam64Id", "name", "preaim",
+             "reactionTime", "accuracy", "accuracyEnemySpotted",
+             "accuracyHead", "shotsFiredEnemySpotted", "shotsFired",
+             "shotsHitEnemySpotted", "shotsHitFriend", "shotsHitFriendHead",
+             "shotsHitFoe", "shotsHitFoeHead", "utilityOnDeathAvg",
+             "heFoesDamageAvg", "heFriendsDamageAvg", "heThrown",
+             "molotovThrown", "smokeThrown", "smokeThrownCT",
+             "smokeThrownCTGood", "smokeThrownCTGoodRatio",
+             "smokeThrownCTFoes", "counterStrafingShotsAll",
+             "counterStrafingShotsBad", "counterStrafingShotsGood",
+             "counterStrafingShotsGoodRatio", "flashbangHitFoe",
+             "flashbangLeadingToKill", "flashbangHitFoeAvgDuration",
+             "flashbangHitFriend", "flashbangThrown", "flashAssist", "score",
+             "initialTeamNumber", "mvps", "ctRoundsWon", "ctRoundsLost",
+             "tRoundsWon", "tRoundsLost", "sprayAccuracy",
+             "molotovFoesDamageAvg", "molotovFriendsDamageAvg", "color",
+             "totalKills", "totalDeaths", "kdRatio", "multi2k", "multi3k",
+             "multi4k", "multi5k", "hltvRating", "hsp", "roundsSurvived",
+             "roundsSurvivedPercentage", "dpr", "totalAssists", "totalDamage",
+             "tradeKillOpportunities", "tradeKillAttempts",
+             "tradeKillsSucceeded", "tradeKillAttemptsPercentage",
+             "tradeKillsSuccessPercentage", "tradeKillOpportunitiesPerRound",
+             "tradedDeathOpportunities", "tradedDeathAttempts",
+             "tradedDeathAttemptsPercentage", "tradedDeathsSucceeded",
+             "tradedDeathsSuccessPercentage",
+             "tradedDeathsOpportunitiesPerRound", "leetifyRating",
+             "personalPerformanceRating", "ctLeetifyRating", "tLeetifyRating",
+             "leetifyUserId", "isCollector", "isProPlan", "isLeetifyStaff"]
+
+    return extract_cols(player_stats, cols)
+
+def cols_from_profile_game(game: dict) -> dict:
+    cols = ["ctLeetifyRating", "ctLeetifyRatingRounds", "dataSource", "elo",
+            "gameFinishedAt", "gameId", "isCs2", "mapName", "matchResult",
+            "scores", "skillLevel", "tLeetifyRating", "tLeetifyRatingRounds",
+            "deaths", "hasBannedPlayer", "kills", "partySize"]
+    return extract_cols(game, cols)
+
+def meta_from_profile(profile: dict) -> dict:
+    meta = profile.get("meta");
+    if not meta:
+        raise Exception("Could not get profile metadata", profile)
+
+    cols = [ "steam64Id", "isCollector", "isLeetifyStaff",
+            "isProPlan", "leetifyUserId", "faceitNickname"]
+
+    return extract_cols(meta, cols)
+
+def insert_value(data: dict[T,U], key: T, value: U) -> dict[T,U]:
+    data[key] = value
+    return data
+
+def convert_game_scores(game: dict) -> dict:
+    score = game.get("scores")
+    if not score:
+        raise Exception("Could not get score from prfile game", game)
+
+    score = score_to_text(score)
+    game["scores"] = score
+    return game
+
+
+
+
+def games_from_profile(profile: dict) -> list:
+    games = profile.get("games")
+    if not games:
+        raise Exception("Could not get games from profile", profile)
+
+    meta = profile.get("meta")
+    if not meta:
+        raise Exception("Could not get profile metadata", profile)
+
+    id = meta.get("leetifyUserId")
+    if not id:
+        raise Exception("Could not get id from profile metadata", meta, profile)
+
+    games = map(cols_from_profile_game, games)
+    games = map(lambda game: insert_value(game, "leetifyUserId", id), games)
+    games = map(convert_game_scores, games)
+    return list(games)
+

src/leetify/__init__.py

@@ -0,0 +1 @@
+from .core import *

src/leetify/core.py

@@ -0,0 +1,20 @@
+import requests
+
+class Leetify:
+    api_base_url = "https://api.leetify.com/api"
+    profile_base_url = f"{api_base_url}/profile"
+    match_base_url = f"{api_base_url}/games"
+
+    def __get_page(self, url: str) -> dict:
+        resp = requests.get(url)
+        return resp.json()
+
+    def get_profile(self, id: str) -> dict:
+        url = f"{self.profile_base_url}/{id}"
+        return self.__get_page(url)
+
+    def get_match(self, id: str) -> dict:
+        url = f"{self.match_base_url}/{id}"
+        return self.__get_page(url)
+
+