forked from Firepup650/FireBot
284 lines
12 KiB
Python
284 lines
12 KiB
Python
#!/usr/bin/python3
|
|
from socket import socket, AF_INET, SOCK_STREAM
|
|
from overrides import bytes, bbytes
|
|
from logs import log
|
|
import re
|
|
from typing import NoReturn
|
|
from config import npbase, servers, __version__
|
|
import commands as cmds
|
|
|
|
|
|
class bot:
|
|
def __init__(self, server: str):
|
|
self.gmode = False
|
|
self.server = server
|
|
self.nicklen = 30
|
|
self.address = servers[server]["address"]
|
|
self.port = servers[server]["port"] if "port" in servers[server] else 6667
|
|
self.channels = servers[server]["channels"]
|
|
self.interval = (
|
|
servers[server]["interval"] if "interval" in servers[server] else 50
|
|
)
|
|
self.nick = "FireBot"
|
|
self.prefix = "."
|
|
self.rebt = "fire"
|
|
self.gblrebt = "all"
|
|
self.adminnames = servers[server]["admins"]
|
|
self.exitcode = f"bye {self.nick.lower()}"
|
|
self.np = re.compile(npbase.replace("MAX", f"{nicklen}"))
|
|
self.queue = []
|
|
self.sock = socket(AF_INET, SOCK_STREAM)
|
|
self.log(f"Start init for {server}")
|
|
|
|
def connect(self) -> None:
|
|
self.log(f"Joining {server}...")
|
|
self.sock.connect((self.address, self.port))
|
|
self.send(f"USER {self.nick} {self.nick} {self.nick} {self.nick}\n")
|
|
self.send(f"NICK {self.nick}\n")
|
|
ircmsg = ""
|
|
while (
|
|
ircmsg.find(f"MODE {self.nick}") == -1
|
|
and ircmsg.find(f"PRIVMSG {self.nick}") == -1
|
|
):
|
|
ircmsg = self.recv().decode()
|
|
if ircmsg != "":
|
|
print(bytes(ircmsg).lazy_decode())
|
|
if ircmsg.find("NICKLEN=") != -1:
|
|
self.nicklen = int(ircmsg.split("NICKLEN=")[1].split(" ")[0])
|
|
self.np = re.compile(npbase.replace("MAX", f"{self.nicklen}"))
|
|
self.log(f"NICKLEN set to {nicklen}")
|
|
elif ircmsg.find("Nickname") != -1:
|
|
self.log("Nickname in use", "WARN")
|
|
self.nick = f"{self.nick}{r.randint(0,1000)}"
|
|
self.send(f"NICK {self.nick}\n")
|
|
self.log(f"nick is now {self.nick}")
|
|
elif ircmsg.startswith("PING "):
|
|
self.ping(ircmsg)
|
|
elif len(ircmsg.split("\x01")) == 3:
|
|
self.CTCPHandler(ircmsg, isRaw=True)
|
|
elif ircmsg.find("Closing Link") != -1:
|
|
self.exit("Closing Link")
|
|
else:
|
|
self.exit("Lost connection to the server")
|
|
self.log(f"Joined {server} successfully!")
|
|
|
|
def join(self, chan: str, origin: str, lock: bool = True) -> None:
|
|
self.log(f"Joining {chan}...")
|
|
chan = chan.replace(" ", "")
|
|
if "," in chan:
|
|
chans = chan.split(",")
|
|
for subchan in chans:
|
|
self.join(subchan, origin)
|
|
return
|
|
if chan.startswith("0") or (chan == "#main" and lock):
|
|
if origin != "null":
|
|
self.sendmsg(f"Refusing to join channel {chan} (protected)", origin)
|
|
return
|
|
if chan in channels and lock:
|
|
if origin != "null":
|
|
self.sendmsg(f"I'm already in {chan}.", origin)
|
|
return
|
|
self.send(f"JOIN {chan}\n")
|
|
while True:
|
|
ircmsg = self.recv().decode()
|
|
if ircmsg != "":
|
|
code = 0
|
|
try:
|
|
code = int(ircmsg.split(" ", 2)[1].strip())
|
|
except (IndexError, ValueError):
|
|
pass
|
|
print(bytes(ircmsg).lazy_decode())
|
|
if ircmsg.startswith("PING "):
|
|
self.ping(ircmsg)
|
|
elif len(ircmsg.split("\x01")) == 3:
|
|
self.CTCP(ircmsg, isRaw=True)
|
|
elif code == 403:
|
|
self.log(f"Joining {chan} failed", "WARN")
|
|
if origin != "null":
|
|
sendmsg(f"{chan} is an invalid channel", origin)
|
|
break
|
|
elif code == 473:
|
|
self.log(f"Joining {chan} failed (+i)", "WARN")
|
|
if origin != "null":
|
|
self.sendmsg(f"{chan} is +i, and I'm not invited.", origin)
|
|
break
|
|
elif code == 520:
|
|
self.log(f"Joining {chan} failed (+O)", "WARN")
|
|
if origin != "null":
|
|
self.sendmsg(f"{chan} is +O, and I'm not an operator.", origin)
|
|
elif code == 366:
|
|
log(f"Joining {chan} succeeded", server)
|
|
if origin != "null":
|
|
self.sendmsg(f"Joined {chan}", origin)
|
|
self.channels[chan] = 0
|
|
break
|
|
|
|
def send(self, command: str) -> int:
|
|
return self.sock.send(bytes(command))
|
|
|
|
def recv(self) -> bytes:
|
|
if self.queue:
|
|
return bytes(self.queue.pop(0))
|
|
data = bytes(self.sock.recv(2048).strip(b"\r\n"))
|
|
if b"\r\n" in data:
|
|
self.queue.extend(data.split(b"\r\n"))
|
|
return bytes(self.queue.pop(0))
|
|
return data
|
|
|
|
def log(self, message: object, level: str = "LOG") -> None:
|
|
log(message, self.server)
|
|
|
|
def exit(self, message: object) -> NoReturn:
|
|
log(message, self.server, "EXIT")
|
|
exit(1)
|
|
|
|
def CTCP(self, msg: str, sender: str = "", isRaw: bool = False) -> bool:
|
|
if isRaw:
|
|
sender = msg.split("!", 1)[0][1:]
|
|
message = msg.split("PRIVMSG", 1)[1].split(":", 1)[1].strip()
|
|
kind = msg.split("\x01")[1].split(" ", 1)[0]
|
|
self.log(f"Responding to CTCP {kind} from {sender}")
|
|
if kind == "VERSION":
|
|
self.notice(
|
|
f"\x01VERSION FireBot {__version__} (https://git.amcforum.wiki/Firepup650/fire-ircbot)\x01",
|
|
sender,
|
|
True,
|
|
)
|
|
return True
|
|
elif kind == "USERINFO":
|
|
self.notice("\x01USERINFO FireBot (Firepup's bot)\x01", sender, True)
|
|
return True
|
|
elif kind == "SOURCE":
|
|
self.notice(
|
|
"\x01SOURCE https://git.amcforum.wiki/Firepup650/fire-ircbot\x01",
|
|
sender,
|
|
True,
|
|
)
|
|
return True
|
|
elif kind == "FINGER":
|
|
self.notice("\x01FINGER Firepup's bot\x01", sender, True)
|
|
return True
|
|
elif kind == "CLIENTINFO":
|
|
self.notice(
|
|
"\x01CLIENTINFO ACTION VERSION USERINFO SOURCE FINGER\x01", sender, True
|
|
)
|
|
return True
|
|
self.log(f"Unknown CTCP {kind}")
|
|
return False
|
|
|
|
def mainloop(self) -> NoReturn:
|
|
self.log("Starting connection..")
|
|
self.connect()
|
|
if "pass" in servers[self.server]:
|
|
self.msg(f"IDENTIFY FireBot {servers[self.server]['pass']}", "NickServ")
|
|
sleep(0.5)
|
|
for chan in self.channels:
|
|
self.join(chan, "null", False)
|
|
while 1:
|
|
raw = self.recv()
|
|
ircmsg = raw.decode()
|
|
if ircmsg == "":
|
|
exit("Probably a netsplit")
|
|
else:
|
|
print(raw.lazy_decode(), sep="\n")
|
|
action = "Unknown"
|
|
try:
|
|
action = ircmsg.split(" ", 2)[1].strip()
|
|
except IndexError:
|
|
pass
|
|
if action == "PRIVMSG":
|
|
# Format of ":[Nick]![ident]@[host|vhost] PRIVMSG [channel] :[message]”
|
|
name = ircmsg.split("!", 1)[0][1:]
|
|
helpErr = False
|
|
if (name.startswith("saxjax") and server == "efnet") or (
|
|
name == "ReplIRC" and server == "replirc"
|
|
):
|
|
if ircmsg.find("<") != -1 and ircmsg.find(">") != -1:
|
|
Nname = ircmsg.split("<", 1)[1].split(">", 1)[0].strip()
|
|
if name == "ReplIRC":
|
|
name = Nname[4:]
|
|
else:
|
|
name = Nname
|
|
message = ircmsg.split(">", 1)[1].strip()
|
|
helpErr = True
|
|
else:
|
|
message = (
|
|
ircmsg.split("PRIVMSG", 1)[1].split(":", 1)[1].strip()
|
|
)
|
|
else:
|
|
message = ircmsg.split("PRIVMSG", 1)[1].split(":", 1)[1].strip()
|
|
if name.endswith("dsc"):
|
|
helpErr = True
|
|
chan = ircmsg.split("PRIVMSG", 1)[1].split(":", 1)[0].strip()
|
|
self.log(
|
|
f'Got "{bytes(message).lazy_decode()}" from "{name}" in "{chan}"',
|
|
)
|
|
if len(name) > nicklen:
|
|
self.log(f"Name too long ({len(name)} > {nicklen})")
|
|
continue
|
|
elif chan not in channels:
|
|
self.log(
|
|
f"Channel not in channels ({chan} not in {channels})",
|
|
"WARN",
|
|
)
|
|
else:
|
|
channels[chan] += 1
|
|
if "goat" in name.lower() and self.gmode == True:
|
|
cmds.goat(self, chan)
|
|
handled = False
|
|
for cmd in cmds.data:
|
|
if mfind(
|
|
message,
|
|
cmd.replace("$BOTNICK", self.nick),
|
|
cmds.data[cmd]["prefix"],
|
|
):
|
|
cmds.call[cmd](self, chan, name)
|
|
handled = True
|
|
break
|
|
if not handled:
|
|
for cmd in cmds.data:
|
|
for alias in cmds.data[cmd]["aliases"]:
|
|
if mfind(
|
|
message,
|
|
alias.replace("$BOTNICK", self.nick),
|
|
cmds.data[cmd]["prefix"],
|
|
):
|
|
cmds.call[cmd](self, chan, name, message)
|
|
handled = True
|
|
break
|
|
if handled:
|
|
break
|
|
if not handled:
|
|
for check in cmds.checks:
|
|
if re.search(
|
|
check.replace("$MAX", self.nicklen).replace(
|
|
"$BOTNICK", self.nick
|
|
),
|
|
message,
|
|
):
|
|
cmds.call[check](self, chan, name, message)
|
|
handled = True
|
|
break
|
|
if not handled and len(message.split("\x01")) == 3:
|
|
if not self.CTCP(message, name):
|
|
CTCP = message.split("\x01")[1]
|
|
if CTCP == "ACTION ducks":
|
|
self.msg("\x01ACTION gets hit by a duck\x01", chan)
|
|
elif CTCP.startswith("ACTION ducks"):
|
|
self.msg(
|
|
f"\x01ACTION gets hit by {CTCP.split(' ', 2)[2]}\x01",
|
|
chan,
|
|
)
|
|
if chan in self.channels and self.channels[chan] >= self.interval:
|
|
r.seed()
|
|
self.channels[chan] = 0
|
|
mm = open("mastermessages.txt", "r")
|
|
q = mm.readlines()
|
|
sel = decode_escapes(
|
|
str(r.sample(q, 1))
|
|
.strip("[]'")
|
|
.replace("\\n", "")
|
|
.strip('"')
|
|
)
|
|
self.msg(f"[QUOTE] {sel}", chan)
|
|
mm.close()
|