Massive hack to automate talk announcements for pgcon 2020. There is DEFINITELY a better way to do this.

pgcon2020_irc_announce,py

#!/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()