hanabi-league/src/fetch_players.py

from dataclasses import dataclass
from typing import List, Optional

import requests_cache
import platformdirs
import csv

import constants
from config import config_manager
import database
from log_setup import logger

session = requests_cache.CachedSession(
    platformdirs.user_cache_dir(constants.APP_NAME) + '/google-docs.requests-cache',
    urls_expire_after={"*": 60 * 5}  # Cache entries for 5 minutes (good especially during development)
)


@dataclass
class Registration:
    user_account: str
    player_name: str
    discord_tag: Optional[str]


def fetch_csv() -> List[str]:
    config = config_manager.get_config()
    url = "https://docs.google.com/spreadsheets/d/{}/gviz/tq?tqx=out:csv".format(config.google_spreadsheet_id)
    response = session.get(url)
    if not response.status_code == 200:
        err_msg = "Failed to fetch signup spreadsheet, url tried was {}".format(url)
        logger.error(err_msg)
        raise ConnectionError(err_msg)
    return response.text.split("\n")


def parse_csv(lines: List[str]):
    reader = csv.reader(lines, delimiter=",", quotechar='"')
    ret = []
    for row in reader:
        if len(row) != 0:
            user_account = row[1]
            player_name = row[2]
            discord_tag = row[3]
            ret.append(Registration(user_account, player_name, discord_tag))
    # We ignore the first row, since that corresponds to the column names
    return ret[1:]


def get_new_registrations():
    data = fetch_csv()
    registrations = sorted(parse_csv(data), key=lambda registration: registration.player_name.lower())
    registered_players = database.get_player_names()

    new_registrations = []

    i = 0  # Index in registrations
    j = 0  # Index in registered players
    # We will traverse both lists at the same time, where we maintain the invariant that j always points to the
    # lowest index whose player name is at least as big as where i points to:
    while i < len(registrations):
        if j == len(registered_players) or registrations[i].player_name.lower() < registered_players[j].lower():
            new_registrations.append(registrations[i])
            i += 1
        elif registrations[i].player_name == registered_players[j]:
            i += 1
        else:
            j += 1
    return new_registrations


def check_account_exists(user_account):
    response = session.get("https://hanab.live/history/{}".format(user_account))
    ok = response.status_code == 200
    return ok


def ask_for_registration_confirmations(registrations: List[Registration], check_existence: bool = True) -> List[Registration]:
    accepted = []
    for registration in registrations:
        if check_existence and not check_account_exists(registration.user_account):
            logger.warning("Rejecting registration {}: Account does not exist on hanab.live".format(registration))
            continue
        response = input("Accept registration account={}, player_name={}, discord_tag={}? [Y/n]".format(
            registration.user_account, registration.player_name, registration.discord_tag
        ))
        if response in ["y", "Y", ""]:
            accepted.append(registration)
            logger.info("Accepted registration".format())
        else:
            logger.info("Rejected registration")
    return accepted


def fetch_players_interactive():
    logger.info("Fetching new players.")
    regs = get_new_registrations()
    if len(regs) == 0:
        logger.info("No new players to register")
        return
    accepted = ask_for_registration_confirmations(regs)
    for registration in accepted:
        database.add_player(registration.user_account, registration.player_name)
    logger.info("Successfully registered {} new players".format(len(accepted)))