From 9bcdd10ed7ac4e662fb61a6d3ac0da2624910fb1 Mon Sep 17 00:00:00 2001 From: Stefan Harmuth Date: Sat, 18 Dec 2021 20:47:45 +0100 Subject: [PATCH] new ircbot interface; still some stuff to figure out --- aocrr_bot.py | 304 +++++++++++++++++++++++++-------------------------- 1 file changed, 149 insertions(+), 155 deletions(-) diff --git a/aocrr_bot.py b/aocrr_bot.py index 0ae7ea2..66984ee 100755 --- a/aocrr_bot.py +++ b/aocrr_bot.py @@ -6,21 +6,20 @@ import sys from datetime import datetime, timedelta from time import sleep from tools.datafiles import JSONFile -from tools.irc import Client, ServerMessage -from tools.schedule import Scheduler +from tools.irc import IrcBot, ServerMessage from tools.tools import human_readable_time_from_delta -#IRC_SERVER = "irc.eu.libera.chat" +# IRC_SERVER = "irc.eu.libera.chat" IRC_SERVER = "irc.tu-ilmenau.de" IRC_PORT = 6667 -#IRC_CHANNEL = "#aocrr" +# IRC_CHANNEL = "#aocrr" IRC_CHANNEL = "#hinterzimmer" IRC_NICK = "aocrr-bot" IRC_USER = "aocrr-bot" IRC_REALNAME = "#aocrr Leaderboard Announcer" -def fetch_leaderboard(session_id: str, year: int = datetime.now().year) -> dict: +def fetch_leaderboard(year: int = datetime.now().year) -> dict: return json.loads( requests.get( "https://adventofcode.com/%d/leaderboard/private/view/711147.json" % year, @@ -29,166 +28,161 @@ def fetch_leaderboard(session_id: str, year: int = datetime.now().year) -> dict: ) -class IrcBot: - def __init__(self): - self.cache = JSONFile("aocrr_bot.cache", create=True) - self.__session_id = open(".session", "r").readlines()[0].strip() - self.irc_client = Client(IRC_SERVER, IRC_PORT, IRC_NICK, IRC_USER, IRC_REALNAME) - self.irc_client.join(IRC_CHANNEL) - self.irc_client.subscribe(ServerMessage.MSG_PRIVMSG, self.on_privmsg) - self.irc_client.subscribe(ServerMessage.RAW, self.on_raw) - self.update_leaderboard() - self.scheduler = Scheduler() - self.scheduler.schedule('leaderboard', timedelta(minutes=15), self.update_leaderboard) +def command_info(msg_from: str, message: str): + ircbot.privmsg( + msg_from, + "I am %s => %s" % (ircbot.getUser().nickname, ircbot.getUser().user) + ) + ircbot.privmsg(msg_from, "I am currently in the following channels:") + for c in ircbot.getChannelList(): + ircbot.privmsg(msg_from, "%s => %s" % (c.name, c.topic)) - def on_privmsg(self, msg_from: str, msg_to: str, message: str): - if msg_to != IRC_CHANNEL: - return - if message.startswith("!info"): - self.irc_client.privmsg( - IRC_CHANNEL, - "I am %s => %s" % (self.irc_client.getUser().nickname, self.irc_client.getUser().user) +def command_today(msg_from: str, message: str): + today = str(datetime.now().day) + day_start = datetime.today().replace(hour=6, minute=0, second=0) + + today_list = [] + for member, member_data in cache.items(): + if not member.startswith("__") and 'days' in member_data and today in member_data['days']: + today_list.append(member) + + ircbot.privmsg( + IRC_CHANNEL, + "Todays leaderboard (last updated: %s):" % cache['__last_update__'] + ) + for i, member in enumerate(sorted(today_list, key=lambda x: cache[x]['days'][today]['score'], reverse=True)): + if i > 3: + sleep(1) # don't flood + + if "1" in cache[member]['days'][today]: + p1_time = "in " + human_readable_time_from_delta( + datetime.fromisoformat(cache[member]['days'][today]['1']) - day_start ) - self.irc_client.privmsg(IRC_CHANNEL, "I am currently in the following channels:") - for c in self.irc_client.getChannelList(): - self.irc_client.privmsg(IRC_CHANNEL, "%s => %s" % (c.name, c.topic)) - elif message.startswith("!today"): - today = str(datetime.now().day) - day_start = datetime.today().replace(hour=6, minute=0, second=0) + else: + p1_time = "not yet solved" - today_list = [] - for member, member_data in self.cache.items(): - if not member.startswith("__") and 'days' in member_data and today in member_data['days']: - today_list.append(member) - - self.irc_client.privmsg( - IRC_CHANNEL, - "Todays leaderboard (last updated: %s):" % self.cache['__last_update__'] + if "2" in cache[member]['days'][today]: + p2_time = "in " + human_readable_time_from_delta( + datetime.fromisoformat(cache[member]['days'][today]['2']) - day_start ) - for i, member in enumerate(sorted(today_list, key=lambda x: self.cache[x]['days'][today]['score'], reverse=True)): - if i > 3: - sleep(1) # don't flood + else: + p2_time = "not yet solved" - if "1" in self.cache[member]['days'][today]: - p1_time = "in " + human_readable_time_from_delta( - datetime.fromisoformat(self.cache[member]['days'][today]['1']) - day_start + ircbot.privmsg( + IRC_CHANNEL, + "%d) %s (Scores: total: %d, today: %d) p1 %s, p2 %s" + % ( + i + 1, + cache[member]['name'], + cache[member]['score'], + cache[member]['days'][today]['score'], + p1_time, + p2_time + ) + ) + + +def command_quit(msg_from: str, message: str): + if msg_from.startswith("stha!") or msg_from.startswith("Pennywise!"): + ircbot.privmsg(msg_from, "Oh, ok ... bye :'(") + ircbot.quit() + sys.exit(0) + + +def on_raw(msg_from: str, msg_type: str, msg_to: str, message: str): + print("[%s] <%s> (%s) -> <%s>: %s" % (datetime.now().strftime("%H:%M:%S"), msg_from, msg_type, msg_to, message)) + + +def calc_scores(): + member_count = len([x for x in cache.keys() if not x.startswith("__")]) + for day in map(str, range(1, 26)): + p1_times = [] + p2_times = [] + for member, member_data in cache.items(): + if member.startswith("__") or day not in member_data['days']: + continue + + cache[member]['days'][day]['score'] = 0 + if '1' in member_data['days'][day]: + p1_times.append(member_data['days'][day]['1']) + if '2' in member_data['days'][day]: + p2_times.append(member_data['days'][day]['2']) + + for member, member_data in cache.items(): + if member.startswith("__") or day not in member_data['days']: + continue + + if '1' in member_data['days'][day] and member_data['days'][day]['1'] in p1_times: + score = member_count - sorted(p1_times).index(member_data['days'][day]['1']) + cache[member]['days'][day]['score'] += score + + if '2' in member_data['days'][day] and member_data['days'][day]['2'] in p2_times: + score = member_count - sorted(p2_times).index(member_data['days'][day]['2']) + cache[member]['days'][day]['score'] += score + + +def update_leaderboard(): + try: + new_leaderboard = fetch_leaderboard() + except: + return # didn't work this time? Well, we'll just try again in 15min ... + + now = datetime.now() + + new_stars = {} + for member, member_data in new_leaderboard['members'].items(): + if member not in cache: + cache[member] = { + 'name': member_data['name'], + 'days': {}, + } + + cache[member]['global_score'] = int(member_data['global_score']) + cache[member]['score'] = int(member_data['local_score']) + cache[member]['stars'] = int(member_data['stars']) + for day in member_data['completion_day_level']: + day_start = datetime(now.year, 12, int(day), 6, 0, 0) + if day not in cache[member]['days']: + cache[member]['days'][day] = {} + + for part in member_data['completion_day_level'][day]: + if part not in cache[member]['days'][day]: + completion_time = datetime.fromtimestamp( + member_data['completion_day_level'][day][part]['get_star_ts'] ) - else: - p1_time = "not yet solved" + if member_data['name'] not in new_stars: + new_stars[member_data['name']] = {} - if "2" in self.cache[member]['days'][today]: - p2_time = "in " + human_readable_time_from_delta( - datetime.fromisoformat(self.cache[member]['days'][today]['2']) - day_start - ) - else: - p2_time = "not yet solved" + finishing_time = human_readable_time_from_delta(completion_time - day_start) + new_stars[member_data['name']]['d' + day + 'p' + part] = finishing_time + cache[member]['days'][day][part] = completion_time.isoformat() - self.irc_client.privmsg( - IRC_CHANNEL, - "%d) %s (Scores: total: %d, today: %d) p1 %s, p2 %s" - % ( - i + 1, - self.cache[member]['name'], - self.cache[member]['score'], - self.cache[member]['days'][today]['score'], - p1_time, - p2_time - ) - ) - elif message.startswith("!quit") and (msg_from.startswith("stha!") or msg_from.startswith("Pennywise!")): - self.irc_client.privmsg(IRC_CHANNEL, "Oh, ok ... bye :'(") - self.irc_client.quit() - sys.exit(0) + if len(new_stars) > 0: + ircbot.privmsg(IRC_CHANNEL, "New Stars found:") + for member, parts in new_stars.items(): + line = member + ": " + line += ", ".join( + "%s (%s)" + % (part, new_stars[member][part]) for part in sorted(new_stars[member].keys()) + ) - def on_raw(self, msg_from: str, msg_type: str, msg_to: str, message: str): - print("[%s] <%s> (%s) -> <%s>: %s" % (datetime.now().strftime("%H:%M:%S"), msg_from, msg_type, msg_to, message)) + ircbot.privmsg(IRC_CHANNEL, line) - def calc_scores(self): - member_count = len([x for x in self.cache.keys() if not x.startswith("__")]) - for day in map(str, range(1, 26)): - p1_times = [] - p2_times = [] - for member, member_data in self.cache.items(): - if member.startswith("__") or day not in member_data['days']: - continue - - self.cache[member]['days'][day]['score'] = 0 - if '1' in member_data['days'][day]: - p1_times.append(member_data['days'][day]['1']) - if '2' in member_data['days'][day]: - p2_times.append(member_data['days'][day]['2']) - - for member, member_data in self.cache.items(): - if member.startswith("__") or day not in member_data['days']: - continue - - if '1' in member_data['days'][day] and member_data['days'][day]['1'] in p1_times: - score = member_count - sorted(p1_times).index(member_data['days'][day]['1']) - self.cache[member]['days'][day]['score'] += score - - if '2' in member_data['days'][day] and member_data['days'][day]['2'] in p2_times: - score = member_count - sorted(p2_times).index(member_data['days'][day]['2']) - self.cache[member]['days'][day]['score'] += score - - def update_leaderboard(self): - try: - new_leaderboard = fetch_leaderboard(self.__session_id) - except: - return # didn't work this time? Well, we'll just try again in 15min ... - - now = datetime.now() - - new_stars = {} - for member, member_data in new_leaderboard['members'].items(): - if member not in self.cache: - self.cache[member] = { - 'name': member_data['name'], - 'days': {}, - } - - self.cache[member]['global_score'] = int(member_data['global_score']) - self.cache[member]['score'] = int(member_data['local_score']) - self.cache[member]['stars'] = int(member_data['stars']) - for day in member_data['completion_day_level']: - day_start = datetime(now.year, 12, int(day), 6, 0, 0) - if day not in self.cache[member]['days']: - self.cache[member]['days'][day] = {} - - for part in member_data['completion_day_level'][day]: - if part not in self.cache[member]['days'][day]: - completion_time = datetime.fromtimestamp( - member_data['completion_day_level'][day][part]['get_star_ts'] - ) - if member_data['name'] not in new_stars: - new_stars[member_data['name']] = {} - - finishing_time = human_readable_time_from_delta(completion_time - day_start) - new_stars[member_data['name']]['d' + day + 'p' + part] = finishing_time - self.cache[member]['days'][day][part] = completion_time.isoformat() - - if len(new_stars) > 0: - self.irc_client.privmsg(IRC_CHANNEL, "New Stars found:") - for member, parts in new_stars.items(): - line = member + ": " - line += ", ".join( - "%s (%s)" - % (part, new_stars[member][part]) for part in sorted(new_stars[member].keys()) - ) - - self.irc_client.privmsg(IRC_CHANNEL, line) - - self.cache['__last_update__'] = datetime.now().isoformat() - self.cache.save() - self.calc_scores() - - def run(self): - while 1: - self.scheduler.run_pending() - self.irc_client.receive() - sleep(0.1) + cache['__last_update__'] = datetime.now().isoformat() + cache.save() + calc_scores() -if __name__ == '__main__': - ircbot = IrcBot() - ircbot.run() +cache = JSONFile("aocrr_bot.cache", create=True) +session_id = open(".session", "r").readlines()[0].strip() + +ircbot = IrcBot(IRC_SERVER, IRC_PORT, IRC_NICK, IRC_USER, IRC_REALNAME) +ircbot.join(IRC_CHANNEL) +ircbot.schedule('update_leaderboard', timedelta(minutes=15), update_leaderboard) +ircbot.on(ServerMessage.RAW, on_raw) +ircbot.register_channel_command("!info", IRC_CHANNEL, command_info) +ircbot.register_channel_command("!today", IRC_CHANNEL, command_today) +ircbot.register_privmsg_command("info", command_info) +ircbot.register_privmsg_command("quit", command_quit) +ircbot.run()