new ircbot interface; still some stuff to figure out
This commit is contained in:
Stefan Harmuth
parent
963cde3ccd
commit
9bcdd10ed7
304
aocrr_bot.py
304
aocrr_bot.py
@ -6,21 +6,20 @@ import sys
|
|||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
from time import sleep
|
from time import sleep
|
||||||
from tools.datafiles import JSONFile
|
from tools.datafiles import JSONFile
|
||||||
from tools.irc import Client, ServerMessage
|
from tools.irc import IrcBot, ServerMessage
|
||||||
from tools.schedule import Scheduler
|
|
||||||
from tools.tools import human_readable_time_from_delta
|
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_SERVER = "irc.tu-ilmenau.de"
|
||||||
IRC_PORT = 6667
|
IRC_PORT = 6667
|
||||||
#IRC_CHANNEL = "#aocrr"
|
# IRC_CHANNEL = "#aocrr"
|
||||||
IRC_CHANNEL = "#hinterzimmer"
|
IRC_CHANNEL = "#hinterzimmer"
|
||||||
IRC_NICK = "aocrr-bot"
|
IRC_NICK = "aocrr-bot"
|
||||||
IRC_USER = "aocrr-bot"
|
IRC_USER = "aocrr-bot"
|
||||||
IRC_REALNAME = "#aocrr Leaderboard Announcer"
|
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(
|
return json.loads(
|
||||||
requests.get(
|
requests.get(
|
||||||
"https://adventofcode.com/%d/leaderboard/private/view/711147.json" % year,
|
"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 command_info(msg_from: str, message: str):
|
||||||
def __init__(self):
|
ircbot.privmsg(
|
||||||
self.cache = JSONFile("aocrr_bot.cache", create=True)
|
msg_from,
|
||||||
self.__session_id = open(".session", "r").readlines()[0].strip()
|
"I am %s => %s" % (ircbot.getUser().nickname, ircbot.getUser().user)
|
||||||
self.irc_client = Client(IRC_SERVER, IRC_PORT, IRC_NICK, IRC_USER, IRC_REALNAME)
|
)
|
||||||
self.irc_client.join(IRC_CHANNEL)
|
ircbot.privmsg(msg_from, "I am currently in the following channels:")
|
||||||
self.irc_client.subscribe(ServerMessage.MSG_PRIVMSG, self.on_privmsg)
|
for c in ircbot.getChannelList():
|
||||||
self.irc_client.subscribe(ServerMessage.RAW, self.on_raw)
|
ircbot.privmsg(msg_from, "%s => %s" % (c.name, c.topic))
|
||||||
self.update_leaderboard()
|
|
||||||
self.scheduler = Scheduler()
|
|
||||||
self.scheduler.schedule('leaderboard', timedelta(minutes=15), self.update_leaderboard)
|
|
||||||
|
|
||||||
def on_privmsg(self, msg_from: str, msg_to: str, message: str):
|
|
||||||
if msg_to != IRC_CHANNEL:
|
|
||||||
return
|
|
||||||
|
|
||||||
if message.startswith("!info"):
|
def command_today(msg_from: str, message: str):
|
||||||
self.irc_client.privmsg(
|
today = str(datetime.now().day)
|
||||||
IRC_CHANNEL,
|
day_start = datetime.today().replace(hour=6, minute=0, second=0)
|
||||||
"I am %s => %s" % (self.irc_client.getUser().nickname, self.irc_client.getUser().user)
|
|
||||||
|
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:")
|
else:
|
||||||
for c in self.irc_client.getChannelList():
|
p1_time = "not yet solved"
|
||||||
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)
|
|
||||||
|
|
||||||
today_list = []
|
if "2" in cache[member]['days'][today]:
|
||||||
for member, member_data in self.cache.items():
|
p2_time = "in " + human_readable_time_from_delta(
|
||||||
if not member.startswith("__") and 'days' in member_data and today in member_data['days']:
|
datetime.fromisoformat(cache[member]['days'][today]['2']) - day_start
|
||||||
today_list.append(member)
|
|
||||||
|
|
||||||
self.irc_client.privmsg(
|
|
||||||
IRC_CHANNEL,
|
|
||||||
"Todays leaderboard (last updated: %s):" % self.cache['__last_update__']
|
|
||||||
)
|
)
|
||||||
for i, member in enumerate(sorted(today_list, key=lambda x: self.cache[x]['days'][today]['score'], reverse=True)):
|
else:
|
||||||
if i > 3:
|
p2_time = "not yet solved"
|
||||||
sleep(1) # don't flood
|
|
||||||
|
|
||||||
if "1" in self.cache[member]['days'][today]:
|
ircbot.privmsg(
|
||||||
p1_time = "in " + human_readable_time_from_delta(
|
IRC_CHANNEL,
|
||||||
datetime.fromisoformat(self.cache[member]['days'][today]['1']) - day_start
|
"%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:
|
if member_data['name'] not in new_stars:
|
||||||
p1_time = "not yet solved"
|
new_stars[member_data['name']] = {}
|
||||||
|
|
||||||
if "2" in self.cache[member]['days'][today]:
|
finishing_time = human_readable_time_from_delta(completion_time - day_start)
|
||||||
p2_time = "in " + human_readable_time_from_delta(
|
new_stars[member_data['name']]['d' + day + 'p' + part] = finishing_time
|
||||||
datetime.fromisoformat(self.cache[member]['days'][today]['2']) - day_start
|
cache[member]['days'][day][part] = completion_time.isoformat()
|
||||||
)
|
|
||||||
else:
|
|
||||||
p2_time = "not yet solved"
|
|
||||||
|
|
||||||
self.irc_client.privmsg(
|
if len(new_stars) > 0:
|
||||||
IRC_CHANNEL,
|
ircbot.privmsg(IRC_CHANNEL, "New Stars found:")
|
||||||
"%d) %s (Scores: total: %d, today: %d) p1 %s, p2 %s"
|
for member, parts in new_stars.items():
|
||||||
% (
|
line = member + ": "
|
||||||
i + 1,
|
line += ", ".join(
|
||||||
self.cache[member]['name'],
|
"%s (%s)"
|
||||||
self.cache[member]['score'],
|
% (part, new_stars[member][part]) for part in sorted(new_stars[member].keys())
|
||||||
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)
|
|
||||||
|
|
||||||
def on_raw(self, msg_from: str, msg_type: str, msg_to: str, message: str):
|
ircbot.privmsg(IRC_CHANNEL, line)
|
||||||
print("[%s] <%s> (%s) -> <%s>: %s" % (datetime.now().strftime("%H:%M:%S"), msg_from, msg_type, msg_to, message))
|
|
||||||
|
|
||||||
def calc_scores(self):
|
cache['__last_update__'] = datetime.now().isoformat()
|
||||||
member_count = len([x for x in self.cache.keys() if not x.startswith("__")])
|
cache.save()
|
||||||
for day in map(str, range(1, 26)):
|
calc_scores()
|
||||||
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)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
cache = JSONFile("aocrr_bot.cache", create=True)
|
||||||
ircbot = IrcBot()
|
session_id = open(".session", "r").readlines()[0].strip()
|
||||||
ircbot.run()
|
|
||||||
|
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()
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user