665 lines
17 KiB
Python
Executable File
665 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
The core bot class for Fulvia.
|
|
"""
|
|
import os
|
|
import sys
|
|
import time
|
|
import threading
|
|
import traceback
|
|
from datetime import datetime
|
|
|
|
from twisted.internet import protocol
|
|
from twisted.words.protocols import irc
|
|
|
|
import db
|
|
import tools
|
|
import config
|
|
import loader
|
|
from trigger import Trigger
|
|
|
|
class Fulvia(irc.IRCClient):
|
|
def __init__(self):
|
|
self.nickname = config.nickname
|
|
self.nick = config.nickname
|
|
"""The bot's current nickname."""
|
|
|
|
self.realname = config.realname
|
|
"""The bot's 'real name', used in whois."""
|
|
|
|
self.username = config.username
|
|
"""The bot's username ident used for logging into the server."""
|
|
|
|
self.host = ""
|
|
"""The bot's host, virtual or otherwise. To be filled in later."""
|
|
|
|
self.static = "static"
|
|
os.makedirs(self.static, exist_ok=True)
|
|
"""The path to the bot's static file directory."""
|
|
|
|
self.log_path = "logs"
|
|
os.makedirs(self.log_path, exist_ok=True)
|
|
"""The path to the bot's log files."""
|
|
|
|
self._log_dump = tools.FulviaMemoryDefault(list)
|
|
"""
|
|
Internal thread-safe dictionary containing log dumps yet to be
|
|
written.
|
|
"""
|
|
|
|
self._last_log_dump = 0
|
|
"""The last time logs were dumped."""
|
|
|
|
self.channels = tools.FulviaMemory()
|
|
"""
|
|
A dictionary of all channels the bot is currently in and the users
|
|
in them.
|
|
"""
|
|
|
|
self.memory = tools.FulviaMemory()
|
|
"""
|
|
A thread-safe general purpose dictionary to be used by whatever
|
|
modules need it.
|
|
"""
|
|
|
|
self.db = db.FulviaDB()
|
|
"""
|
|
A class with some basic interactions for the bot's sqlite3 databse.
|
|
"""
|
|
|
|
self._hooks = []
|
|
"""
|
|
A list containing all function names to be hooked with every message
|
|
received.
|
|
"""
|
|
|
|
self.commands = {}
|
|
"""
|
|
A dictionary containing all commands to look for and the name
|
|
of the function they call.
|
|
"""
|
|
|
|
self._times = {}
|
|
"""
|
|
A dictionary full of times when certain functions were called for
|
|
use with the @rate decorator. Keys are function names.
|
|
"""
|
|
|
|
self.url_callbacks = {}
|
|
"""
|
|
A dictionary of URL callbacks to allow other modules interact with
|
|
the URL module.
|
|
"""
|
|
|
|
self._user_joined = []
|
|
"""These get called when a user joins a channel."""
|
|
|
|
self._disabled_modules = config.disabled_modules
|
|
"""These modules will NOT be loaded when load_modules() is called."""
|
|
|
|
self.load_modules()
|
|
|
|
|
|
def load_modules(self):
|
|
"""
|
|
Find and load all of our modules.
|
|
"""
|
|
print(f"Loading modules...")
|
|
self._hooks = []
|
|
self.commands = {}
|
|
self._times = {}
|
|
self.url_callbacks = {}
|
|
# ensure they're empty
|
|
|
|
modules = loader.find_modules()
|
|
loaded = 0
|
|
failed = 0
|
|
for name, path in modules.items():
|
|
try:
|
|
if name in self._disabled_modules:
|
|
continue
|
|
loader.load_module(self, path)
|
|
loaded += 1
|
|
except Exception as e:
|
|
print(f"{name} failed to load.")
|
|
print(e)
|
|
traceback.print_exc()
|
|
failed += 1
|
|
continue
|
|
print(f"Registered {loaded} modules,")
|
|
print(f"{failed} modules failed to load")
|
|
|
|
|
|
def register_callable(self, func):
|
|
"""
|
|
Registers all callable functions loaded from modules into one
|
|
convenient table.
|
|
"""
|
|
if hasattr(func, 'commands'):
|
|
for cmd in func.commands:
|
|
self.commands[cmd] = func
|
|
|
|
if func.hook:
|
|
self._hooks.append(func)
|
|
|
|
if func.rate or func.channel_rate or func.global_rate:
|
|
self._times[func.__name__] = {}
|
|
|
|
if hasattr(func, 'url_callback'):
|
|
for url in func.url_callback:
|
|
self.url_callbacks[url] = func
|
|
|
|
if func.user_joined:
|
|
self._user_joined.append(func)
|
|
|
|
|
|
def unregister_callable(self, func):
|
|
"""
|
|
Removes the provided function object from the internal list of
|
|
module functions.
|
|
"""
|
|
if not callable(func) or not loader.is_triggerable(func):
|
|
return
|
|
|
|
if hasattr(func, 'commands'):
|
|
for command in func.commands:
|
|
self.commands.pop(command)
|
|
|
|
if func.hook:
|
|
self._hooks.remove(func.__name__)
|
|
|
|
if func.rate or func.channel_rate or func.global_rate:
|
|
self._times.pop(func.__name__)
|
|
|
|
if hasattr(func, 'url_callback'):
|
|
for url in func.url_callback:
|
|
self.url_callbacks.pop(url)
|
|
|
|
if func.user_joined:
|
|
self._user_joined.remove(func)
|
|
|
|
|
|
def stillConnected(self):
|
|
"""Returns true if the bot is still connected to the server."""
|
|
if self._heartbeat:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
|
|
def call(self, func, bot, trigger):
|
|
"""
|
|
A wrapper for calling module functions so that we can have nicer
|
|
error handling.
|
|
"""
|
|
try:
|
|
func(bot, trigger)
|
|
except Exception as e:
|
|
msg = type(e).__name__ + ": " + str(e)
|
|
self.msg(trigger.channel, msg)
|
|
traceback.print_exc()
|
|
|
|
|
|
def log(self, channel, text):
|
|
"""
|
|
Logs text to file. Logging structure is
|
|
{self.log_path}/{self.server}/{channel}/{date}.log
|
|
|
|
Only permits log dumping once per second.
|
|
"""
|
|
# TODO: use time module instead of datetime
|
|
t = datetime.fromtimestamp(time.time())
|
|
timestamp = t.strftime(config.default_time_format)
|
|
self._log_dump[channel].append(timestamp + " " + text)
|
|
|
|
if time.time() - self._last_log_dump > 1:
|
|
for ch in self._log_dump.keys():
|
|
fname = t.strftime("%Y-%m-%d") + ".log"
|
|
path = os.path.join(self.log_path, self.hostname, ch)
|
|
os.makedirs(path, exist_ok=True)
|
|
with open(os.path.join(path, fname), "a+") as file:
|
|
file.write("\n".join(self._log_dump[ch]) + "\n")
|
|
|
|
del(self._log_dump)
|
|
self._log_dump = tools.FulviaMemoryDefault(list)
|
|
|
|
|
|
## Actions involving the bot directly.
|
|
## These will get called automatically.
|
|
|
|
def privmsg(self, user, channel, message):
|
|
"""
|
|
Called when the bot receives a PRIVMSG, which can come from channels
|
|
or users alike.
|
|
"""
|
|
nick = user.partition("!")[0].partition("@")[0]
|
|
if channel.startswith("#"):
|
|
opSym = self.channels[channel].users[nick].op_level
|
|
else:
|
|
opSym = ''
|
|
channel = nick
|
|
line = f"<{opSym}{nick}> {message}"
|
|
self.log(channel, line)
|
|
|
|
funcs = []
|
|
if message.startswith(config.prefix) and message != config.prefix:
|
|
command = message.partition(" ")[0]
|
|
command = command.replace(config.prefix, "", 1)
|
|
cmd = self.commands.get(command)
|
|
if not cmd:
|
|
return
|
|
funcs.append(cmd)
|
|
|
|
funcs += self._hooks
|
|
|
|
for func in funcs:
|
|
trigger = Trigger(user, channel, message, "PRIVMSG")
|
|
bot = FulviaWrapper(self, trigger)
|
|
|
|
if func.rate:
|
|
t = self._times[func_name].get(trigger.nick, 0)
|
|
if time.time() - t < func.rate and not trigger.admin:
|
|
return
|
|
self._times[func_name][trigger.nick] = time.time()
|
|
|
|
if func.channel_rate:
|
|
t = self._times[func_name].get(trigger.channel, 0)
|
|
if time.time() - t < func.channel_rate and not trigger.admin:
|
|
return
|
|
self._times[func_name][trigger.channel] = time.time()
|
|
|
|
if func.global_rate:
|
|
t = self._times[func_name].get("global", 0)
|
|
if time.time() - t < func.channel_rate and not trigger.admin:
|
|
return
|
|
self._times[func_name]["global"] = time.time()
|
|
|
|
if func.thread == True:
|
|
t = threading.Thread(target=self.call,args=(func, bot, trigger))
|
|
t.start()
|
|
else:
|
|
self.call(func, bot, trigger)
|
|
|
|
|
|
def joined(self, channel):
|
|
"""Called when the bot joins a new channel."""
|
|
line = f"-!- {self.nickname} [{self.username}@{self.host}] "
|
|
line += f"has joined {channel}"
|
|
self.log(channel, line)
|
|
|
|
print(f"Joined {channel}")
|
|
self.channels[channel] = tools.Channel(channel)
|
|
user = tools.User(self.nickname)
|
|
self.channels[channel].users[self.nickname] = user
|
|
|
|
|
|
def left(self, channel, reason=""):
|
|
"""Called when the bot leaves a channel."""
|
|
line = f"-!- {self.nickname} [{self.username}@{self.host}] "
|
|
line += f"has left {channel} [{reason}]"
|
|
self.log(channel, line)
|
|
|
|
print(f"Parted {channel}")
|
|
self.channels.pop(channel)
|
|
|
|
|
|
def noticed(self, user, channel, message):
|
|
"""Called when the bot receives a NOTICE from a user or channel."""
|
|
# TODO: something?
|
|
pass
|
|
|
|
|
|
def modeChanged(self, user, channel, add_mode, modes, args):
|
|
"""Called when users or channel's modes are changed."""
|
|
line = f"-!- mode/{channel} ["
|
|
if add_mode:
|
|
line += "+"
|
|
else:
|
|
line += "-"
|
|
line += modes
|
|
if any(args):
|
|
line += " " + " ".join(args)
|
|
line += "] by " + user.partition("!")[0].partition("@")[0]
|
|
self.log(channel, line)
|
|
|
|
if not channel.startswith("#"): # server level
|
|
return
|
|
|
|
for n, mode in enumerate(modes):
|
|
if args[n] == None: # channel mode
|
|
if add_mode:
|
|
self.channels[channel].modes.add(mode)
|
|
else:
|
|
self.channels[channel].modes.remove(mode)
|
|
|
|
elif mode in tools.op_level.keys(): # user mode, op_level mode
|
|
nick = args[n]
|
|
user = self.channels[channel].users[nick]
|
|
if add_mode:
|
|
user.op_level += mode
|
|
else:
|
|
user.op_level.replace(mode, '', 1)
|
|
|
|
else: # user mode, non-op_level mode
|
|
continue
|
|
|
|
|
|
def signedOn(self):
|
|
"""Called when the bot successfully connects to the server."""
|
|
if config.oper_password:
|
|
self.sendLine("OPER " + config.nickname + ' ' + config.oper_password)
|
|
print(f"Signed on as {self.nickname}")
|
|
self.whois(self.nickname)
|
|
|
|
line = f"*** Signed onto {self.hostname} as "
|
|
line += f"{self.nickname}!{self.username}@{self.host}"
|
|
self.log(self.hostname, line)
|
|
|
|
for channel in config.channels:
|
|
self.join(channel)
|
|
|
|
|
|
def kickedFrom(self, channel, kicker, message):
|
|
"""Called when the bot is kicked from a channel."""
|
|
line = f"-!- {self.nickname} was kicked from {channel} "
|
|
line += f"by {kicker} [{message}]"
|
|
self.log(channel, line)
|
|
|
|
self.channels.pop(channel)
|
|
|
|
|
|
def nickChanged(self, nick):
|
|
"""Called when the bot changes it's nickname."""
|
|
line = f"-!- you are now known as {nick}"
|
|
|
|
for channel_name, channel in self.channels.items():
|
|
self.log(channel_name, line)
|
|
|
|
user = channel.users.pop(self.nickname)
|
|
user.nick = nick
|
|
channel.users[nick] = user
|
|
|
|
|
|
## Actions the bot observes other users doing in the channel.
|
|
## These will get called automatically.
|
|
|
|
def userJoined(self, user, channel):
|
|
"""Called when the bot sees another user join a channel."""
|
|
nick, _, host = user.partition("!")
|
|
|
|
line = f"-!- {nick} [{host}] has joined {channel}"
|
|
self.log(channel, line)
|
|
|
|
new_user = tools.User(nick)
|
|
self.channels[channel].users[nick] = new_user
|
|
|
|
for func in self._user_joined:
|
|
trigger = Trigger(user, channel, f"{user} has joined", "PRIVMSG")
|
|
bot = FulviaWrapper(self, trigger)
|
|
t = threading.Thread(target=self.call,args=(func, bot, trigger))
|
|
t.start()
|
|
|
|
|
|
def userLeft(self, user, channel, reason=""):
|
|
"""Called when the bot sees another user leave a channel."""
|
|
nick, _, host = user.partition("!")
|
|
|
|
line = f"-!- {nick} [{host}] has left {channel} [{reason}]"
|
|
self.log(channel, line)
|
|
|
|
self.channels[channel].users.pop(nick)
|
|
|
|
|
|
def userQuit(self, user, quitMessage):
|
|
"""Called when the bot sees another user disconnect from the network."""
|
|
nick, _, host = user.partition("!")
|
|
line = f"-!- {nick} [{host}] has quit [{quitMessage}]"
|
|
|
|
for channel_name, channel in self.channels.items():
|
|
if not nick in channel.users:
|
|
continue
|
|
self.log(channel_name, line)
|
|
|
|
channel.users.pop(nick)
|
|
|
|
|
|
def userKicked(self, kickee, channel, kicker, message):
|
|
"""
|
|
Called when the bot sees another user getting kicked from the channel.
|
|
"""
|
|
line =f"-!- {kickee} was kicked from {channel} by {kicker} [{message}]"
|
|
self.log(channel, line)
|
|
|
|
self.channels[channel].users.pop(kickee)
|
|
|
|
|
|
def topicUpdated(self, user, channel, newTopic):
|
|
"""Called when the bot sees a user update the channel's topic."""
|
|
line = f"-!- {user} changed the topic of {channel} to: {newTopic}"
|
|
self.log(channel, line)
|
|
|
|
self.channels[channel].topic = newTopic
|
|
|
|
|
|
def userRenamed(self, oldname, newname):
|
|
"""Called when the bot sees a user change their nickname."""
|
|
line = "-!- {oldname} is now known as {newname}"
|
|
|
|
for channel_name, channel in self.channels.items():
|
|
self.log(channel_name, line)
|
|
|
|
user = channel.users.pop(oldname)
|
|
user.nick = newname
|
|
channel.users[newname] = user
|
|
|
|
|
|
def namesReceived(self, channel, channel_type, nicklist):
|
|
"""
|
|
Called when the bot receives a list of names in the channel from the
|
|
server.
|
|
"""
|
|
self.channels[channel].channel_type = channel_type
|
|
for nick in nicklist:
|
|
op_level = ''
|
|
if nick[0] in tools.op_level.keys():
|
|
op_level = nick[0]
|
|
nick = nick[1:]
|
|
user = tools.User(nick)
|
|
user.op_level = op_level
|
|
self.channels[channel].users[nick] = user
|
|
|
|
|
|
def whoisUser(self, nick, ident, host, realname):
|
|
"""
|
|
Called when the bot receives a WHOIS about a particular user.
|
|
"""
|
|
if nick == self.nickname:
|
|
self.username = ident
|
|
self.host = host
|
|
self.realname = realname
|
|
else:
|
|
for channel_name, channel in self.channels.items():
|
|
if nick in channel:
|
|
user = channel[nick]
|
|
user.ident = ident
|
|
user.host = host
|
|
user.realname = realname
|
|
|
|
|
|
def whoisIdle(self, nick, idle, signon):
|
|
"""
|
|
Called when the bot receives a WHOIS about a particular user.
|
|
nick - nickname of the user
|
|
idle - seconds since last activity from user
|
|
signon - unix timestamp when user signed on
|
|
"""
|
|
if self.memory.get("idle_callbacks"):
|
|
self.memory["idle_callbacks"][nick].callback((nick, idle, signon))
|
|
self.memory["idle_callbacks"].pop(nick)
|
|
|
|
|
|
## User commands, from client->server
|
|
|
|
def msg(self, user, message, length=None):
|
|
"""
|
|
Sends a message 'message' to 'user' (can be a user or a channel).
|
|
If 'length' is specified, the message will be split into lengths
|
|
of that size. If 'length' is not specified, the bot will determine
|
|
an appropriate length automatically.
|
|
"""
|
|
if user.startswith("#"):
|
|
opSym = self.channels[user].users[self.nickname].op_level
|
|
else:
|
|
opSym = ''
|
|
line = f"<{opSym}{self.nickname}> {message}"
|
|
self.log(user, line)
|
|
|
|
super().msg(user, message, length)
|
|
|
|
|
|
def reply(self, text, dest, reply_to, notice=False):
|
|
"""
|
|
Sends a message to 'dest' prefixed with 'reply_to' and a colon,
|
|
ala "reply_to: text". More useful with the wrapper class.
|
|
"""
|
|
text = reply_to + ": " + text
|
|
self.msg(dest, text)
|
|
|
|
|
|
def quit(self, message=""):
|
|
"""
|
|
Disconnects from quitting normally results in the factory
|
|
reconnecting us, so we need to include shutting down the factory.
|
|
"""
|
|
print("Quit from server")
|
|
irc.IRCClient.quit(self, message)
|
|
self.factory.stopTrying()
|
|
sys.exit(0) # why doesn't this work?
|
|
|
|
|
|
## Raw server->client messages
|
|
|
|
def irc_RPL_NAMREPLY(self, prefix, params):
|
|
"""
|
|
Called when we have received a list of names in the channel from the
|
|
server.
|
|
"""
|
|
channel_types = {"@": "secret", "*": "private", "=": "public"}
|
|
channel_type = channel_types[params[1]]
|
|
channel = params[2]
|
|
nicklist = params[3].split()
|
|
self.namesReceived(channel, channel_type, nicklist)
|
|
|
|
|
|
def irc_JOIN(self, prefix, params):
|
|
"""
|
|
Called when a user joins a channel.
|
|
"""
|
|
nick = prefix.split('!')[0]
|
|
channel = params[-1]
|
|
if nick == self.nickname:
|
|
self.joined(channel)
|
|
else:
|
|
self.userJoined(prefix, channel)
|
|
|
|
|
|
def irc_PART(self, prefix, params):
|
|
"""
|
|
Called when a user leaves a channel.
|
|
"""
|
|
nick = prefix.split('!')[0]
|
|
channel = params[0]
|
|
if len(params) > 1:
|
|
reason = params[1]
|
|
else:
|
|
reason = ""
|
|
|
|
if nick == self.nickname:
|
|
self.left(channel, reason)
|
|
else:
|
|
self.userLeft(prefix, channel, reason)
|
|
|
|
|
|
def irc_QUIT(self, prefix, params):
|
|
"""
|
|
Called when a user has quit.
|
|
"""
|
|
nick = prefix.split('!')[0]
|
|
self.userQuit(prefix, params[0])
|
|
|
|
|
|
def irc_RPL_WHOISUSER(self, prefix, params):
|
|
"""
|
|
Called when we receive the server's response from our "WHOIS [user]"
|
|
query. Contains hostmask information.
|
|
"""
|
|
_, nick, ident, host, _, realname = params
|
|
self.whoisUser(nick, ident, host, realname)
|
|
|
|
|
|
def irc_RPL_WHOISIDLE(self, prefix, params):
|
|
"""
|
|
Called when we receive the server's response from our "WHOIS [user]"
|
|
query. Contains idle and signon time information.
|
|
"""
|
|
_, nick, idle, signon, _ = params
|
|
self.whoisIdle(nick, idle, signon)
|
|
|
|
|
|
def lineReceived(self, line):
|
|
"""Overridden to fix broken unicode issues."""
|
|
if bytes != str and isinstance(line, bytes):
|
|
line = line.decode('utf-8', errors='surrogateescape')
|
|
super().lineReceived(line)
|
|
|
|
|
|
class FulviaWrapper():
|
|
"""
|
|
A wrapper class for Fulvia to provide default destinations for msg
|
|
methods and so forth.
|
|
"""
|
|
def __init__(self, fulvia, trigger):
|
|
object.__setattr__(self, '_bot', fulvia)
|
|
object.__setattr__(self, '_trigger', trigger)
|
|
# directly setting these values would cause a recursion loop
|
|
# overflow
|
|
|
|
def __getattr__(self, attr):
|
|
"""Redirects getting attributes to the master bot instance."""
|
|
return getattr(self._bot, attr)
|
|
|
|
def __setattr__(self, attr, value):
|
|
"""Redirects setting attributes to the master bot instance."""
|
|
return setattr(self._bot, attr, value)
|
|
|
|
def msg(self, user=None, message=None, length=None):
|
|
"""
|
|
If user is None it will default to the channel the message was
|
|
received on.
|
|
"""
|
|
if user == None:
|
|
raise TypeError("msg() missing 2 required positional " \
|
|
+ "arguments: 'user' and 'message'")
|
|
if message == None:
|
|
# assume the first argument is the message
|
|
message = user
|
|
user = self._trigger.channel
|
|
self._bot.msg(user, message, length)
|
|
|
|
def reply(self, message, destination=None, reply_to=None, notice=False):
|
|
"""
|
|
If destination is None, it defaults to the channel the message
|
|
was received on.
|
|
If reply_to is None, it defaults to the person who sent the
|
|
message.
|
|
"""
|
|
if destination is None:
|
|
destination = self._trigger.channel
|
|
if reply_to is None:
|
|
reply_to = self._trigger.nick
|
|
self._bot.reply(message, destination, reply_to, notice)
|
|
|
|
|
|
class FulviaFactory(protocol.ReconnectingClientFactory):
|
|
protocol = Fulvia
|