nebula/chatterbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bunch of refactoring i guess

Browse Source
This commit is contained in:
= 2025-04-04 06:54:31 +00:00
parent ea79b23b28
commit 5f0c0b2182
2 changed files with 165 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

329
bot.py
View File

@ -3,8 +3,8 @@ from time import sleep, time
from json import load, dump from json import load, dump
import re import re
channel_re = re.compile(r"PRIVMSG (#*\w+)") privmsg_channel_re = re.compile(r"PRIVMSG (#*\w+)")
name_re = re.compile(r"^:([^!]*)!") nick_re = re.compile(r"^:([^!]*)!")
# timeout = 86400 # 24 hours # timeout = 86400 # 24 hours
timeout = 15 timeout = 15
@ -15,171 +15,172 @@ helptext = "i am a bot by ~nebula. i try to make it easier for users to discover
helptext_short = "see https://git.tilde.town/nebula/chatterbot for instructions" helptext_short = "see https://git.tilde.town/nebula/chatterbot for instructions"
class IRCBot(): class IRCBot():
def __init__(self): def __init__(self):
try: try:
with open("config.json", "r") as f: with open("config.json", "r") as f:
self.config = load(f) self.state = load(f)
except FileNotFoundError: except FileNotFoundError:
exit("no config.json") exit("no config.json")
self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.s.connect((host, port)) self.s.connect((host, port))
self.nick = self.config["nick"] self.nick = self.state["nick"]
self.sendline(f"NICK {self.nick}") self.send_raw_line(f"NICK {self.nick}")
self.sendline(f"USER {self.nick} 0 * :{self.config['realname']}") self.send_raw_line(f"USER {self.nick} 0 * :{self.state['realname']}")
for channel in self.config["channels"]: for channel in self.state["channels"]:
self.sendline(f"JOIN {channel}") self.send_raw_line(f"JOIN {channel}")
self.commands = [ self.commands = [
("invite", self.invite), ("invite", self.invite),
("kick", self.kick) ("kick", self.kick)
] ]
def write_config(self):
with open("config.json", "w") as f:
dump(self.config, f, 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)
del self.config["times"][channel]
del self.config["counts"][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 = self.config["times"][channel] = time()
self.write_config()
return this_time
def set_time(self, channel, this_time): def write_state(self):
self.config["times"][channel] = this_time with open("state.json", "w") as f:
dump(self.state, f, indent=2)
def counter(self, channel): def join_channel(self, channel):
try: self.send_raw_line(f"JOIN {channel}")
self.config["counts"][channel] += 1 self.state["channels"].append(channel)
value = self.config["counts"][channel] self.write_state()
except KeyError:
value = self.config["counts"][channel] = 1
self.write_config()
return value
def reset_count(self, channel):
self.config["counts"][channel] = 0
self.write_config()
def command_loop(self): def part_channel(self, channel):
while True: if channel in ("#tildetown", "#bots"):
char = self.s.recv(1) self.send_raw_line(f"i will not leave {channel}. want to block me? see https://git.tilde.town/nebula/chatterbot")
if not char: return
exit(f"{self.nick}: no response from IRC server") self.send_raw_line(f"PART {channel}")
line = b"" self.state["channels"].remove(channel)
while char != b"\n": del self.state["times"][channel]
if char != b"\r": del self.state["counts"][channel]
line += char self.write_state()
char = self.s.recv(1)
line = line.decode("UTF-8").strip() def send_raw_line(self, line):
if line.startswith("PING"): if line:
pong = "PONG " + line[5:] return self.s.send(bytes(f"{line}\r\n", "UTF-8"))
self.sendline(pong) return
continue
channel_search = channel_re.search(line) def send(self, channel, content):
if not channel_search: if isinstance(content, list):
continue for line in content:
channel = channel_search.group(1) self.send_raw_line(f"PRIVMSG {channel} :{line}")
name_search = name_re.search(line) sleep(0.5)
if name_search: elif isinstance(content, str):
name = name_search.group(1) self.send_raw_line(f"PRIVMSG {channel} :{content}")
else:
name = None def help(_, __):
if name and not channel.startswith("#"): return helptext
channel = name
try: def invite(self, _, arguments):
message_body = line[line.index(" :") + 2:] if not arguments:
except (IndexError, ValueError): return helptext_short
message_body = "" lines = []
if message_body: for channel in arguments:
if message_body.startswith("!rollcall"): if not channel.startswith("#"):
self.send(channel, helptext) lines.append("channel name must start with #")
continue continue
elif message_body.startswith("!chatterbot"): elif channel in self.state["channels"]:
arguments = message_body.strip().lower()[11:] lines.append(f"i am already in {channel}!")
if not arguments: else:
self.send(channel, helptext) self.join_channel(channel)
continue lines.append(f"i have (allegedly) joined {channel}")
arguments = arguments.split() return lines
for command, callback in self.commands:
if command not in arguments: def kick(self, channel, arguments):
continue if not arguments:
self.send(channel, callback(channel, arguments[1:])) self.part_channel(channel)
else: return
if channel in ("#tildetown", "#bots"): for channel in arguments:
continue self.part_channel(channel)
channel_time = self.check_time(channel)
now = time() def check_time(self, channel):
count = self.counter(channel) try:
delta = now - channel_time this_time = self.state["times"][channel]
if delta > timeout and count < messages_within_timeout: except KeyError:
self.reset_count(channel) this_time = self.state["times"][channel] = time()
self.send("#bots", f"i hear activity in {channel}...") self.write_state()
elif count < messages_within_timeout and delta > timeout: return this_time
self.reset_count
self.set_time(channel, now) def set_time(self, channel, this_time):
self.state["times"][channel] = this_time
def counter(self, channel):
try:
self.state["counts"][channel] += 1
value = self.state["counts"][channel]
except KeyError:
value = self.state["counts"][channel] = 1
self.write_state()
return value
def reset_count(self, channel):
self.state["counts"][channel] = 0
self.write_state()
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.send_raw_line(pong)
continue
privmsg_channel_search = privmsg_channel_re.search(line)
if not privmsg_channel_search:
continue
channel = privmsg_channel_search.group(1)
nick_search = nick_re.search(line)
if nick_search:
nick = nick_search.group(1)
else:
nick = None
if nick and not channel.startswith("#"):
channel = nick
try:
message_body = line[line.index(" :") + 2:]
except (IndexError, ValueError):
message_body = ""
if message_body:
if message_body.startswith("!rollcall") or message_body.startswith("!help"):
self.send(channel, helptext)
continue
elif message_body.startswith("!chatterbot"):
arguments = message_body.strip().lower()[11:]
if not arguments:
self.send(channel, helptext)
continue
arguments = arguments.split()
for command, callback in self.commands:
if command not in arguments:
continue
self.send(channel, callback(channel, arguments[1:]))
else:
if channel in ("#tildetown", "#bots"):
continue
# i have not figured out this part yet
# channel_time = self.check_time(channel)
# now = time()
# count = self.counter(channel)
# delta = now - channel_time
# if delta > timeout:
# if count < messages_within_timeout:
# self.send("#bots", f"i hear activity in {channel}...")
# self.reset_count(channel)
# elif and channel_time :
# self.reset_count
# self.set_time(channel, now)
if __name__ == "__main__": if __name__ == "__main__":
bot = IRCBot() bot = IRCBot()
try: try:
bot.command_loop() bot.command_loop()
except KeyboardInterrupt: except KeyboardInterrupt:
exit() exit()

8
config.json.example
View File

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