hanabi-league/src/render_site.py

from pathlib import Path
from typing import Dict, List

import jinja2
import datetime
import psycopg2.extras

import constants
import utils
import ratings
from dataclasses import dataclass

from database import conn_manager


@dataclass
class PlayerEntry:
    player_name: str
    user_accounts: str
    score: int


@dataclass
class Leader:
    title: str
    entry: PlayerEntry


@dataclass
class VariantStats:
    name: str
    num_players: int
    rating: int
    games_played: int
    games_won: int


def get_rating_lists() -> Dict[int, List[PlayerEntry]]:
    cur = conn_manager.get_connection().cursor()
    rating_types = [utils.get_rating_type(x) for x in [False, True]]
    leaderboard = {
        utils.get_rating_type(x): []
        for x in [True, False]
    }
    for rating_type in rating_types:
        cur.execute(
            "SELECT * FROM ("
            "  SELECT DISTINCT ON (user_accounts.user_id)"
            "    player_name,"
            "    string_agg(user_accounts.username, %s ORDER BY user_accounts.username) AS user_accounts,"
            "    COALESCE(value_after, rating) AS current_rating"
            "    FROM users "
            "    LEFT OUTER JOIN user_ratings "
            "      ON user_ratings.user_id = users.id AND user_ratings.type = %s "
            "    LEFT OUTER JOIN user_accounts "
            "      ON users.id = user_accounts.user_id "
            "    INNER JOIN user_base_ratings "
            "      ON users.id = user_base_ratings.user_id AND user_base_ratings.type = %s"
            "  GROUP BY (user_accounts.user_id, player_name, value_after, league_id, rating) "
            "    ORDER BY user_accounts.user_id, league_id DESC"
            "  ) AS ratings "
            "ORDER BY current_rating DESC",
            (", ", rating_type, rating_type)
        )
        for (player_name, user_accounts, current_rating) in cur.fetchall():
            leaderboard[rating_type].append(PlayerEntry(player_name, user_accounts, round(current_rating)))

    return leaderboard


def get_leaders(rating_lists: Dict, streak_lists: Dict) -> Dict[int, Dict[str, Leader]]:
    leaders = {}
    for rating_type, rating_list in rating_lists.items():
        if len(rating_list) != 0:
            leader = rating_list[0]
            leaders[rating_type] = {
                'Player Rating': Leader('Highest Rating', leader)
            }
    for rating_type, streak_list in streak_lists.items():
        if len(streak_list) != 0:
            leader = streak_list[0]
            if rating_type not in leaders.keys():
                leaders[rating_type] = {}
            leaders[rating_type]['Top Streak'] = Leader('Longest Win Streak', leader)
    return leaders


def get_streak_lists():
    cur = conn_manager.get_connection().cursor(cursor_factory=psycopg2.extras.DictCursor)
    rating_types = [utils.get_rating_type(x) for x in [False, True]]
    leaderboard = {
        utils.get_rating_type(x): []
        for x in [True, False]
    }
    for rating_type in rating_types:
        cur.execute(
            "SELECT * FROM ("
            "  SELECT "
            "      player_name,"
            "      string_agg(user_accounts.username, %s ORDER BY user_accounts.username) AS user_accounts,"
            "      COALESCE(maximum_streak, %s) AS maximum_streak"
            "    FROM users "
            "    LEFT OUTER JOIN user_statistics"
            "      ON users.id = user_statistics.user_id AND variant_type = %s"
            "    LEFT OUTER JOIN user_accounts "
            "      ON users.id = user_accounts.user_id "
            "    GROUP BY (user_accounts.user_id, player_name, maximum_streak) "
            "  ) AS streaks "
            "ORDER BY maximum_streak DESC",
            (", ", 0, rating_type)
        )
        for (player_name, user_accounts, maximum_streak) in cur.fetchall():
            leaderboard[rating_type].append(PlayerEntry(player_name, user_accounts, maximum_streak))
    return leaderboard


def get_variant_ratings():
    cur = conn_manager.get_new_cursor()
    cur.execute(
        "SELECT"
        "  ratings.name,"
        "  ratings.num_players,"
        "  current_rating,"
        "  COUNT(games.id) AS games_played,"
        "  COUNT(games.id) FILTER (WHERE ratings.num_suits * 5 = games.score) AS games_won "
        "FROM "
        "  ("
        "  SELECT DISTINCT ON (variants.id, variant_base_ratings.num_players)"
        "    variants.id,"
        "    variants.num_suits,"
        "    name,"
        "    variant_base_ratings.num_players,"
        "    COALESCE(variant_ratings.value_after, variant_base_ratings.rating) AS current_rating "
        "  FROM variants "
        "  LEFT OUTER JOIN variant_base_ratings"
        "    ON variants.id = variant_base_ratings.variant_id "
        "  LEFT OUTER JOIN variant_ratings "
        "    ON variant_ratings.variant_id = variant_base_ratings.variant_id AND variant_ratings.num_players = variant_base_ratings.num_players "
        "  GROUP BY (variants.id, name, variant_base_ratings.num_players, variant_base_ratings.rating, variant_ratings.league_id, variant_ratings.value_after) "
        "  ORDER BY variants.id, variant_base_ratings.num_players, league_id DESC"
        "  ) AS ratings "
        "LEFT OUTER JOIN games "
        "  ON games.variant_id = ratings.id AND games.num_players = ratings.num_players "
        "GROUP BY (ratings.id, ratings.name, ratings.num_players, ratings.current_rating)"
        "ORDER BY (ratings.id, ratings.num_players)"
        ""
    )
    return [
        VariantStats(variant_name, num_players, rating, games_played, games_won)
        for (variant_name, num_players, rating, games_played, games_won) in cur.fetchall()
    ]


def get_total_games():
    cur = conn_manager.get_new_cursor()
    cur.execute("SELECT COUNT(league_id) FROM games")
    (num_games,) = cur.fetchone()
    return num_games


def get_num_players():
    cur = conn_manager.get_new_cursor()
    cur.execute("SELECT COUNT(id) FROM users")
    (num_users,) = cur.fetchone()
    return num_users


def render_leaderboard():
    rating_lists = get_rating_lists()
    streak_lists = get_streak_lists()
    leaders = get_leaders(rating_lists, streak_lists)

    leaderboards = {
        'Player Rating': rating_lists,
        'Maximum Streak': streak_lists
    }

    env = jinja2.Environment(loader=jinja2.FileSystemLoader('templates'))
    template = env.get_template('content.html')
    # rendered_html = template.render(leaders=leaders, leaderboards=leaderboards, variants=variants)
    rendered_html = template.render(
        leaders=leaders,
        leaderboards=leaderboards,
        total_games_played=get_total_games(),
        total_players=get_num_players(),
        latest_run=datetime.datetime.now().isoformat(),
        variants=get_variant_ratings()
        #        variants=variants,
    )
    output_file = Path(constants.WEBSITE_OUTPUT_DIRECTORY) / 'index.html'
    output_file.parent.mkdir(exist_ok=True, parents=True)
    with open(output_file, 'w') as f:
        f.write(rendered_html)


render_leaderboard()