Skip to content

Instantly share code, notes, and snippets.

@rwarren
Last active May 28, 2020 18:41
Show Gist options
  • Save rwarren/0860453ff2c49f146cef725868575f9c to your computer and use it in GitHub Desktop.
Save rwarren/0860453ff2c49f146cef725868575f9c to your computer and use it in GitHub Desktop.
Massive hack to automate talk announcements for pgcon 2020. There is DEFINITELY a better way to do this.
#!/usr/bin/env python
# THIS FILE IS A MAAAAAAAAAAAASSIVE HACK QUICKLY MADE FOR THE ONLINE PGCON_2020
# There is DEFINITELY a better way to do this.
import argparse
import datetime
import sched
import socket
import sys
import time
import typing as th
UTC_OFFSET = datetime.datetime.utcnow().timestamp() - time.time()
PING_DELAY_s = 120
PONG_DELAY_s = 2
def utcnow_s() -> float:
return time.time() + UTC_OFFSET
def dlog(msg: str) -> None:
sys.stdout.write(f"{msg}\n")
sys.stdout.flush() # for nohup buffering
class Done(Exception): pass
class IrcBot:
"A cheap hack for sending messages to irc channels."
_sock: socket.socket
_channel: str
def __init__(self,
host: str,
port: int,
#nick: str,
ident: str,
#realname: str,
channel: str,
) -> None:
self._channel = channel
self._sock = socket.socket()
self._sock.connect((host, port))
self._sock.setblocking(False)
self._send(f"USER {ident} 0 * :{ident}")
self._send(f"NICK {ident}")
self._send(f"JOIN {channel}")
def _send(self, s: str) -> None:
self._sock.send(s.encode("utf8") + b"\r\n")
self.clear_recv_buf()
def clear_recv_buf(self):
"""Eat any bytes in the recv buffer, and repsond to PING requests."""
try:
buf: bytes = self._sock.recv(9999999)
lines = buf.decode("utf8").split("\r\n")
for line in lines:
if line.strip():
dlog(line)
if line.startswith("PING"):
self._send(f"PONG {line.split(' ', 1)[1]}")
dlog(buf.decode("utf8"))
except:
pass # nonblocking hack
def send_msg(self, msg: str) -> None:
self._send(f"PRIVMSG {self._channel} :{msg}")
def send_ping(self):
dlog(f"PINGing {self._channel}")
self._send(f"PING :foo")
class Talk:
start_time_utc: th.Union[None, datetime.datetime]
summary: str
url: str
irc_chan: str
def __init__(self, src_fp: th.TextIO) -> None:
# ical is simple enough to parse with a hack
self.start_time_utc = None
self.summary = ""
self.url = ""
self.irc_chan = ""
for line in src_fp:
line = line.strip()
if line.startswith("END:VEVENT"):
break
elif line.startswith("DTSTART:"):
self.start_time_utc = datetime.datetime.strptime(line, "DTSTART:%Y%m%dT%H%M%SZ")
elif line.startswith("SUMMARY:"):
self.summary = line[8:]
elif line.startswith("LOCATION:"):
self.irc_chan = f"#pgcon-stream{line[-1]}"
elif line.startswith("URL:"):
self.url = line[4:].replace("10.80.0.70", "www.pgcon.org")
else:
raise Done()
def __str__(self) -> str:
return f"{self.start_time_utc}: {self.irc_chan} -- {self.summary}"
def load_schedule(ical_path: str) -> th.List[Talk]:
dlog("Loading schedule")
talks = []
with open(ical_path, "r") as fp:
while True:
try:
talks.append(Talk(fp))
except Done:
break
return talks
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("-i", action="store",
help="Path to the ical file with the schedule.")
return parser.parse_args()
def do_announce(irc_bots: th.Dict[str, IrcBot], talk: Talk) -> None:
dlog(f"Announcing: {talk}")
bot = irc_bots[talk.irc_chan]
bot.send_msg(f"***********************")
bot.send_msg(f"******* A talk is starting in 2 minutes:")
bot.send_msg(f"******* Summary: {talk.summary}")
bot.send_msg(f"******* URL: {talk.url}")
bot.send_msg(f"***********************")
def run_announcer(irc_bots: th.Dict[str, IrcBot], talks: th.List[Talk]) -> None:
s = sched.scheduler(utcnow_s, time.sleep)
#s.enterabs(utcnow_s() + 1, 1, do_announce, (cmdline_args, talks[5]))
#s.enterabs(utcnow_s() + 2, 1, do_announce, (cmdline_args, talks[6]))
#s.enterabs(utcnow_s() + 3, 1, do_announce, (cmdline_args, talks[7]))
def ping_forever():
for bot in irc_bots.values():
bot.send_ping()
s.enter(PING_DELAY_s, 1, ping_forever)
def pong_forever():
for bot in irc_bots.values():
bot.clear_recv_buf()
s.enter(PONG_DELAY_s, 1, pong_forever)
ping_forever()
pong_forever()
start_s = utcnow_s()
for talk in talks:
talk_start_s = talk.start_time_utc.timestamp()
if talk_start_s < start_s:
# don't announce stuff that has already happened
continue
dlog(f"Scheduling {talk}")
s.enterabs(talk.start_time_utc.timestamp() - 120, 1, do_announce, (irc_bots, talk, ))
dlog("Running scheduler (time to wait!)")
s.run()
def main():
irc_bots: th.List[IrcBot]
cmdline_args = parse_args()
talks = load_schedule(cmdline_args.i)
channels = {talk.irc_chan for talk in talks}
irc_bots = {channel: IrcBot("irc.freenode.net", 6667, f"pgcon_bot{channel[-1]}", channel)
for channel in channels}
#for irc_bot in irc_bots.values():
# irc_bot.send_msg("Testing 123")
run_announcer(irc_bots, talks)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment