initial commit

bot.py

@@ -0,0 +1,168 @@
+import socket
+from time import sleep, time
+from json import load, dump
+import re
+
+channel_re = re.compile(r"PRIVMSG (#*\w+)")
+name_re = re.compile(r"^:([^!]*)!")
+
+timeout = 86400 # 24 hours
+messages_within_24h = 100
+host = "localhost"
+port = 6667
+helptext = "i am a bot by ~nebula. i try to make it easier for users to discover new and more obscure channels on irc. instructions for use (or to block me from showing messages in your client) are available at https://git.tilde.town/nebula/chatterbot"
+helptext_short = "see https://git.tilde.town/nebula/chatterbot for instructions" 
+
+class IRCBot():
+    def __init__(self):
+        try:
+            with open("config.json", "r") as f:
+                self.config = load(f)
+        except FileNotFoundError:
+            exit("no config.json")
+        self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.s.connect((host, port))
+        self.nick = self.config["nick"]
+        self.sendline(f"NICK {self.nick}")
+        self.sendline(f"USER {self.nick} 0 * :{self.config["realname"]}")
+        for channel in self.config["channels"]:
+            self.join_channel(channel)
+        self.commands = [
+            ("invite", self.invite)
+            ("kick", self.kick)
+        ]
+        
+    def write_config(self):
+        with open("config.json", "w") as f:
+            dump(self.config, indent=2)
+
+    def join_channel(self, channel):
+        self.sendline(f"JOIN {channel}")
+        self.config["channels"].append(channel)
+        self.write_config()
+
+    def part_channel(self, channel):
+        if channel == "#tildetown":
+            self.sendline("i will not leave #tildetown. want to block me? see https://git.tilde.town/nebula/chatterbot")
+            return
+        elif channel == "#bots":
+            self.sendline("i will not leave #bots. want to block me? see https://git.tilde.town/nebula/chatterbot")
+            return
+        self.sendline(f"PART {channel}")
+        self.config["channels"].remove(channel)
+        self.write_config()
+
+    def sendline(self, line):
+        if line:
+            return self.s.send(bytes(f"{line}\r\n", "UTF-8"))
+        return None
+
+    def send(self, channel, content):
+        if isinstance(content, list):
+            for line in content:
+                self.sendline(f"PRIVMSG {channel} :{line}")
+                sleep(0.5)
+        elif isinstance(content, str):
+            self.sendline(f"PRIVMSG {channel} :{content}")
+
+    def help(_, __):
+        return helptext
+
+    def invite(self, _, arguments):
+        if not arguments:
+            return helptext_short
+        lines = []
+        for channel in arguments:
+            if not channel.startswith("#"):
+                lines.append("channel name must start with #")
+                continue
+            elif channel in self.config["channels"]:
+                lines.append(f"i am already in {channel}!")
+            else:
+                self.join_channel(channel)
+                lines.append(f"i have (allegedly) joined {channel}")
+        return lines
+
+    def kick(self, channel, arguments):
+        if not arguments:
+            self.part_channel(channel)
+            return None
+        for channel in arguments:
+            self.part_channel(channel)
+
+    def check_time(self, channel):
+        try:
+            this_time = self.config["times"][channel]
+        except KeyError:
+            this_time = time()
+            self.config["times"][channel] = this_time
+            self.write_config()
+        return this_time
+    
+    def set_time(self, channel, this_time):
+        self.config["times"][channel] = this_time
+
+    def counter(self, channel):
+        try:
+            self.config["counts"][channel] += 1
+            value = self.config["counts"][channel]
+        except KeyError:
+            value = self.config["counts"][channel] = 1
+        self.write_config()
+        return value
+
+    def command_loop(self):
+        while True:
+            char = self.s.recv(1)
+            if not char:
+                exit(f"{self.nick}: no response from IRC server")
+            line = b""
+            while char != b"\n":
+                if char != b"\r":
+                    line += char
+                char = self.s.recv(1)
+            line = line.decode("UTF-8").strip()
+            if line.startswith("PING"):
+                pong = "PONG " + line[5:]
+                self.sendline(pong)
+                continue
+            channel_search = channel_re.search(line)
+            if not channel_search:
+                continue
+            channel = channel_search.group(1)
+            name_search = name_re.search(line)
+            if name_search:
+                name = name_search.group(1)
+            else:
+                name = None
+            if name and not channel.startswith("#"):
+                channel = name
+            try:
+                message_body = line[line.index(" :") + 2:]
+            except (IndexError, ValueError):
+                message_body = ""
+            if message_body:
+                if message_body.startswith("!rollcall"):
+                    self.send(helptext)
+                    continue
+                elif message_body.startswith("!chatterbot"):
+                    arguments = message_body.strip().lower()[11:]
+                    if not arguments:
+                        self.send(helptext)
+                        continue
+                    arguments = arguments.split()
+                    for command, callback in self.commands:
+                        if command not in arguments:
+                            continue
+                        self.send(callback(channel, arguments[1:]))
+                else:
+                    if channel in ("#tildetown", "#bots"):
+                        continue
+                    channel_time = self.check_time(channel)
+                    now = time()
+                    if self.counter(channel) > messages_within_24h and now - channel_time > timeout:
+                        self.send("#tildetown", f"i hear activity in {channel}...")
+                    self.set_time(channel, now)
+                    self.config["counts"][channel] = 0
+                    
+

config.json

@@ -0,0 +1,8 @@
+{
+    "nick": "chatterbot",
+    "realname": "a bot by ~nebula",
+    "channels": ["#bots"],
+    "times": {},
+    "counts": {}
+}
+  

config.json.example

@@ -0,0 +1,8 @@
+{
+    "nick": "chatterbot",
+    "realname": "a bot by ~nebula",
+    "channels": ["#bots"],
+    "times": {},
+    "counts": {}
+}
+