leetify-data/src/extract.py

from .leetify import Leetify
from os import environ as env
from .transform import (
    meta_from_profile,
    games_from_profile,
    player_stats_from_game,
)
import psycopg2 as pg
from typing import Tuple


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 vals_in_order(data: dict, keys: list):
    vals = [data[key] for key in keys]
    return vals


def dict_to_col_val(data: dict) -> Tuple[list, list]:
    cols = []
    vals = []
    for k, v in data.items():
        cols.append(k)
        vals.append(v)

    return cols, vals


def dict_to_sql_str(data: dict) -> Tuple[str, str]:
    cols, vals = dict_to_col_val(data)
    vals = map(str, vals)

    cols = ", ".join(cols)
    vals = ", ".join(vals)

    return cols, vals


def extract_profile_meta(id, api=Leetify(), conn=None):
    if not conn:
        conn = connect()

    cols = [
        "name",
        "steam64Id",
        "isCollector",
        "isLeetifyStaff",
        "isProPlan",
        "leetifyUserId",
        "faceitNickname",
    ]
    profile = api.get_profile(id)
    meta = meta_from_profile(profile)
    vals = vals_in_order(meta, cols)

    with conn.cursor() as curs:
        sql = f"""
        INSERT INTO data.profile_meta ({', '.join(cols)})
        VALUES ({', '.join(['%s'] * len(vals))});
        """
        curs.execute(sql, vals)
    conn.commit()


def extract_profile_games(profile, conn=None):
    if not conn:
        conn = connect()

    cols = [
        "leetifyUserId",
        "ctLeetifyRating",
        "ctLeetifyRatingRounds",
        "dataSource",
        "elo",
        "gameFinishedAt",
        "gameId",
        "isCs2",
        "mapName",
        "matchResult",
        "scores",
        "skillLevel",
        "tLeetifyRating",
        "tLeetifyRatingRounds",
        "deaths",
        "hasBannedPlayer",
        "kills",
        "partySize",
    ]

    games = games_from_profile(profile)
    games = map(lambda game: vals_in_order(game, cols), games)

    with conn.cursor() as curs:
        curs.executemany(
            f"""
            INSERT into data.profile_game ({', '.join(cols)})
            VALUES ({', '.join(['%s'] * len(cols))});
            """,
            games,
        )
    conn.commit()


def extract_player_stats(game, conn=None):
    if not conn:
        conn = connect()
    stats = player_stats_from_game(game)

    games = map(dict_to_sql_str, stats)

    with conn.cursor() as curs:
        for (
            cols,
            vals,
        ) in games:
            curs.execute(
                """INSERT INTO data.player_stats (%s)
                              VALUES (%s);""",
                (cols, vals),
            )
    conn.commit()