hanabi-league/ratings.py

87 lines
3.1 KiB
Python

from typing import List, Dict
import utils
from database import conn_manager
from log_setup import logger
def next_game_to_rate():
cur = conn_manager.get_new_cursor()
cur.execute("SELECT games.id FROM games "
"LEFT OUTER JOIN user_ratings"
" ON games.league_id = user_ratings.league_id "
"WHERE user_ratings.league_id IS NULL "
"ORDER BY games.league_id ASC "
"LIMIT 1"
)
query_result = cur.fetchone()
if query_result is None:
return
(game_id,) = query_result
return game_id
def get_current_ratings(user_ids: List[int], rating_type: int) -> Dict[int, float]:
"""
Fetches the current ratings for specified players and rating type from DB
@return: Mapping user_id -> current rating
"""
cur = conn_manager.get_new_cursor()
cur.execute("SELECT user_id, rating FROM user_base_ratings "
"WHERE user_id IN ({}) AND type = %s".format(", ".join("%s" for _ in user_ids)),
user_ids + [rating_type]
)
base_ratings = cur.fetchall()
cur.execute("SELECT user_ratings.user_id, value_after FROM user_ratings "
"INNER JOIN ("
" SELECT user_id, type, MAX(league_id) AS max_league_id"
" FROM user_ratings "
" GROUP BY (user_id, type)"
" ) AS latest_user_ratings "
" ON user_ratings.league_id = latest_user_ratings.max_league_id "
)
current_ratings = cur.fetchall()
ratings: Dict[int, float] = {}
for user_id, base_rating in base_ratings:
ratings[user_id] = base_rating
for user_id, rating in current_ratings:
ratings[user_id] = rating
return ratings
def process_rating_of_next_game():
game_id = next_game_to_rate()
if game_id is None:
logger.info("All games already processed for rating changes.")
return
logger.verbose("Processing rating for game {}".format)
cur = conn_manager.get_new_cursor()
cur.execute("SELECT games.league_id, games.num_players, games.score, variants.num_suits, variants.clue_starved "
"FROM games "
"INNER JOIN variants "
" ON games.variant_id = variants.id "
"WHERE games.id = %s",
(game_id,)
)
league_id, num_players, score, num_suits, clue_starved = cur.fetchone()
cur.execute("SELECT game_participants.user_id FROM games "
"INNER JOIN game_participants "
" ON games.id = game_participants.game_id "
"WHERE games.id = %s",
(game_id,)
)
user_ids = cur.fetchall()
if len(user_ids) != num_players:
err_msg = "Player number mismatch: Expected {} participants for game {}, but only found {} in DB: [{}]".format(
num_players, game_id, len(user_ids), ", ".join(user_ids)
)
logger.error(err_msg)
raise ValueError(err_msg)
rating_type = utils.get_rating_type(clue_starved)
ratings = get_current_ratings(user_ids, rating_type)