From ea789d6eab45e105bff3021cbf5f077e10a14401 Mon Sep 17 00:00:00 2001 From: Antoine Stevan Date: Sat, 10 Jul 2021 17:24:26 +0200 Subject: [PATCH] Fully commented and documented. --- src/bot.py | 139 +++++++++++++++++++++++++++++++++++++++-------------- 1 file changed, 103 insertions(+), 36 deletions(-) diff --git a/src/bot.py b/src/bot.py index 4597cfb..0e9bcf1 100644 --- a/src/bot.py +++ b/src/bot.py @@ -11,14 +11,14 @@ def __init__(self, game, payload, needs_flush=False): Args ---- game : str - the name of the game and thus the score board. + the name of the game and thus the scoreboard. payload : [(int, str, str), ...] the payload to send to the server, namely [(score, date, username), ...]. needs_flush : bool, optional tells whether to flush the channel or not. """ super(ScoreBot, self).__init__() - # the game name and thus the score board name. + # the game name and thus the scoreboard name. self.game = game self.head = f"**{self.game}**:" @@ -29,10 +29,15 @@ def __init__(self, game, payload, needs_flush=False): # the channel id the bot has access to. self.channel_id = 862644543207637042 + # internal variables. + self.sep = " | " + self.lines = 10 + self.cols = (4, 5, 19) + async def on_ready(self): """ Runs when the bot connects to the server. Its only purpose is to read the channel, create or modify the game - score board and send it to the channel. + scoreboard and send it to the channel. Args ---- @@ -45,70 +50,132 @@ async def on_ready(self): print(log_msg) print('-' * len(log_msg)) - # send the score board to the server. - await self.send_score_board() + # send the scoreboard to the server. + await self.send_scoreboard() # close the bot, its job is done. await self.close() - async def flush(self): - hist = await self.get_channel(self.channel_id).history(limit=None).flatten() + async def _flush(self): + """ + Flushes all the history of the channel. + Args + ---- - for i, msg in enumerate(hist): + Returns + ------- + history : list + the list of all previous message still present in the channel after the flush. + """ + # retrieve history and flush it if needed. + history = await self.get_channel(self.channel_id).history(limit=None).flatten() + print("previous channel history.") + for i, msg in enumerate(history): print(f"{msg.content}") if self.needs_flush: await msg.delete() + # update the history if a flush happened. if self.needs_flush: - hist = await self.get_channel(self.channel_id).history(limit=None).flatten() + history = await self.get_channel(self.channel_id).history(limit=None).flatten() + + return history - return hist + def _extract_payload(self, scoreboard): + """ + Extracts all the scores from an existing scoreboard and adds the current payload. + + Args + ---- + scoreboard : str + the scoreboard. It is simply a multi-lines string where each row, except the head of the scoreboard, + contains information about the score performed. + + Returns + ------- + payload : [(int, str, str), ...] + the list of all the previous recorded scores, namely [(score, date, username), ...]. + """ + previous_scoreboard = scoreboard.split('\n') # split the lines. - def extract_previous_payload(self, score_board, sep): - previous_score_board = score_board.content.split('\n') - lines = list(map(lambda x: x.replace('`', '').split(sep)[1:], previous_score_board[3:])) - lines = list(map(lambda line: map(lambda el: el.replace('`', '').strip(), line), lines)) + self.payload + # skip the first 3 lines + remove the '`', split with the separator and remove the first column. + lines = list(map(lambda x: x.replace('`', '').split(self.sep)[1:], previous_scoreboard[3:])) + # strip everything in the sub lists and append the current payload. + lines = list(map(lambda line: map(lambda el: el.strip(), line), lines)) + self.payload - scores, dates, names = tuple(zip(*lines)) - scores = list(map(int, scores)) - payload = list(zip(scores, dates, names)) + scores, dates, names = tuple(zip(*lines)) # extract the information. + scores = list(map(float, scores)) # convert the scores to numbers. + payload = list(zip(scores, dates, names)) # recombine the information. return payload - async def build_score_board(self, payload, sep): - # payload = list(zip(*payload)) + async def _build_scoreboard(self, payload): + """ + Builds a valid scoreboard from a payload. + + Args + ---- + payload : [(int, str, str), ...] + the list of all the previous recorded scores, namely [(score, date, username), ...]. + + Returns + ------- + scoreboard : str + the scoreboard. It is simply a multi-lines string where each row, except the head of the scoreboard, + contains information about the score performed. + """ + # sort the payload from biggest to smallest score, i.e. with the score which is ine first place. + print(payload) payload.sort() payload = payload[::-1] - cols = (4, 5, 19, max(list(map(len, list(zip(*payload))[1])))) + # the width of the columns, the last one adapts to the usernames, which are unknown a priori. + cols = self.cols + (max(list(map(len, list(zip(*payload))[1]))),) - lines = [f"\t`{'rank':^{cols[0]}}{sep}{'score':^{cols[1]}}{sep}{'date':^{cols[2]}}{sep}{'name':^{cols[3]}}`"] - lines += ["\t`" + "-+-".join(['-' * col for col in cols]) + '`'] - lines += [f"\t`{i + 1:>{cols[0]}d}{sep}{score:>{cols[1]}}{sep}{date:>{cols[2]}}{sep}{name:>{cols[3]}}`" for - i, (score, date, name) in enumerate(payload)] + # the first line with the names of the columns. + lines = [self.sep.join([f"\t`{'rank':^{cols[0]}}", f"{'score':^{cols[1]}}", + f"{'date':^{cols[2]}}", f"{'name':^{cols[3]}}`"])] + lines += ["\t`" + "-+-".join(['-' * col for col in cols]) + '`'] # a separation line. + # the rest of the lines, with the scores and more info. + lines += [self.sep.join([f"\t`{i + 1:>{cols[0]}d}", f"{score:>{cols[1]}}", f"{date:>{cols[2]}}", + f"{name:>{cols[3]}}`"]) for i, (score, date, name) in enumerate(payload)] - score_board = '\n'.join([self.head] + lines[:10]) - return score_board + scoreboard = '\n'.join([self.head] + lines[:self.lines + 2]) # simply join the head and enough lines with '\n'. + return scoreboard - async def send_score_board(self): - history = await self.flush() + async def send_scoreboard(self): + """ + The main method of the bot. Reads the history of the channels, extracts scores from previous valid + scoreboards, creates or modifies the scoreboard with new scores and send it back to the server. + + Args + ---- + + Returns + ------- + """ + # read the history. + history = await self._flush() - score_boards = [msg for msg in history if msg.content.startswith(self.head)] + # isolate scoreboards of the game. + scoreboards = [msg for msg in history if msg.content.startswith(self.head)] - sep = ' | ' - if len(history) == 0 or len(score_boards) == 0: - print(f"creating score board for {self.game}") + # use the whole current payload to create a scoreboard from scratch. + if len(history) == 0 or len(scoreboards) == 0: + print(f"creating scoreboard for {self.game}") payload = self.payload + # simply extracts the previous scores and add the current ones. also deletes all invalid scoreboards. else: - payload = self.extract_previous_payload(score_boards[0], sep) - for msg in score_boards: + payload = self._extract_payload(scoreboards[0].content) + for msg in scoreboards: await msg.delete() print("appending to", self.game) - score_board = await self.build_score_board(payload=payload, sep=sep) - await self.get_channel(self.channel_id).send(score_board) + # build and send the scoreboard. + scoreboard = await self._build_scoreboard(payload=payload) + await self.get_channel(self.channel_id).send(scoreboard) if __name__ == '__main__':