104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
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)))
|