reset line indentation back to 4
This commit is contained in:
=
parent
ad7164edb7
commit
0d4b658d1c
324
bot.py
324
bot.py
@ -15,172 +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.state = 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.state["nick"]
|
self.nick = self.state["nick"]
|
||||||
self.send_raw_line(f"NICK {self.nick}")
|
self.send_raw_line(f"NICK {self.nick}")
|
||||||
self.send_raw_line(f"USER {self.nick} 0 * :{self.state['realname']}")
|
self.send_raw_line(f"USER {self.nick} 0 * :{self.state['realname']}")
|
||||||
for channel in self.state["channels"]:
|
for channel in self.state["channels"]:
|
||||||
self.send_raw_line(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_state(self):
|
||||||
|
with open("state.json", "w") as f:
|
||||||
|
dump(self.state, f, indent=2)
|
||||||
|
|
||||||
|
def join_channel(self, channel):
|
||||||
|
self.send_raw_line(f"JOIN {channel}")
|
||||||
|
self.state["channels"].append(channel)
|
||||||
|
self.write_state()
|
||||||
|
|
||||||
|
def part_channel(self, channel):
|
||||||
|
if channel in ("#tildetown", "#bots"):
|
||||||
|
self.send_raw_line(f"i will not leave {channel}. want to block me? see https://git.tilde.town/nebula/chatterbot")
|
||||||
|
return
|
||||||
|
self.send_raw_line(f"PART {channel}")
|
||||||
|
self.state["channels"].remove(channel)
|
||||||
|
del self.state["times"][channel]
|
||||||
|
del self.state["counts"][channel]
|
||||||
|
self.write_state()
|
||||||
|
|
||||||
|
def send_raw_line(self, line):
|
||||||
|
if line:
|
||||||
|
return self.s.send(bytes(f"{line}\r\n", "UTF-8"))
|
||||||
|
return
|
||||||
|
|
||||||
|
def send(self, channel, content):
|
||||||
|
if isinstance(content, list):
|
||||||
|
for line in content:
|
||||||
|
self.send_raw_line(f"PRIVMSG {channel} :{line}")
|
||||||
|
sleep(0.5)
|
||||||
|
elif isinstance(content, str):
|
||||||
|
self.send_raw_line(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.state["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
|
||||||
|
for channel in arguments:
|
||||||
|
self.part_channel(channel)
|
||||||
|
|
||||||
|
def check_time(self, channel):
|
||||||
|
try:
|
||||||
|
this_time = self.state["times"][channel]
|
||||||
|
except KeyError:
|
||||||
|
this_time = self.state["times"][channel] = time()
|
||||||
|
self.write_state()
|
||||||
|
return this_time
|
||||||
|
|
||||||
def write_state(self):
|
def set_time(self, channel, this_time):
|
||||||
with open("state.json", "w") as f:
|
self.state["times"][channel] = this_time
|
||||||
dump(self.state, f, indent=2)
|
|
||||||
|
|
||||||
def join_channel(self, channel):
|
def counter(self, channel):
|
||||||
self.send_raw_line(f"JOIN {channel}")
|
try:
|
||||||
self.state["channels"].append(channel)
|
self.state["counts"][channel] += 1
|
||||||
self.write_state()
|
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 part_channel(self, channel):
|
def command_loop(self):
|
||||||
if channel in ("#tildetown", "#bots"):
|
while True:
|
||||||
self.send_raw_line(f"i will not leave {channel}. want to block me? see https://git.tilde.town/nebula/chatterbot")
|
char = self.s.recv(1)
|
||||||
return
|
if not char:
|
||||||
self.send_raw_line(f"PART {channel}")
|
exit(f"{self.nick}: no response from IRC server")
|
||||||
self.state["channels"].remove(channel)
|
line = b""
|
||||||
del self.state["times"][channel]
|
while char != b"\n":
|
||||||
del self.state["counts"][channel]
|
if char != b"\r":
|
||||||
self.write_state()
|
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
|
||||||
|
|
||||||
def send_raw_line(self, line):
|
# channel_time = self.check_time(channel)
|
||||||
if line:
|
# now = time()
|
||||||
return self.s.send(bytes(f"{line}\r\n", "UTF-8"))
|
# count = self.counter(channel)
|
||||||
return
|
# delta = now - channel_time
|
||||||
|
# if delta > timeout:
|
||||||
def send(self, channel, content):
|
# if count < messages_within_timeout:
|
||||||
if isinstance(content, list):
|
# self.send("#bots", f"i hear activity in {channel}...")
|
||||||
for line in content:
|
# self.reset_count(channel)
|
||||||
self.send_raw_line(f"PRIVMSG {channel} :{line}")
|
|
||||||
sleep(0.5)
|
# elif and channel_time :
|
||||||
elif isinstance(content, str):
|
# self.reset_count
|
||||||
self.send_raw_line(f"PRIVMSG {channel} :{content}")
|
# self.set_time(channel, now)
|
||||||
|
|
||||||
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.state["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
|
|
||||||
for channel in arguments:
|
|
||||||
self.part_channel(channel)
|
|
||||||
|
|
||||||
def check_time(self, channel):
|
|
||||||
try:
|
|
||||||
this_time = self.state["times"][channel]
|
|
||||||
except KeyError:
|
|
||||||
this_time = self.state["times"][channel] = time()
|
|
||||||
self.write_state()
|
|
||||||
return this_time
|
|
||||||
|
|
||||||
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()
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user