fulvia/bot.py

#!/usr/bin/env python3
"""
The core bot class for Fulvia.
"""
import os
import sys
import time
import functools
import threading
import traceback

from twisted.internet import protocol
from twisted.words.protocols import irc

import db
import tools
import loader
from trigger import Trigger

class Fulvia(irc.IRCClient):
	def __init__(self, config):
		self.config = config
		"""The bot's config, loaded from file."""

		self.nickname = config.core.nickname
		self.nick = config.core.nickname
		"""The bot's current nickname."""

		self.realname = config.core.realname
		"""The bot's 'real name', used in whois."""

		self.username = config.core.username
		"""The bot's username ident used for logging into the server."""

		self.prefix = config.core.prefix
		"""The command prefix the bot watches for."""

		self.static = os.path.join(config.homedir, "static")
		"""The path to the bot's static file directory."""

		self.channels = tools.FulviaMemory()
		"""
		A dictionary of all channels the bot is currently in and the users
		in them.
		"""

		self.users = tools.FulviaMemory()
		"""A dictionary of all users the bot is aware of."""

		self.memory = tools.FulviaMemory()
		"""
		A thread-safe general purpose dictionary to be used by whatever
		modules need it.
		"""

		self.db = db.FulviaDB(self.config)
		"""
		A class with some basic interactions for the bot's sqlite3 databse.
		"""

		self._callables = {}
		"""
		A dictionary containing all callable functions loaded from 
		modules. Keys are the functions name.
		"""

		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.load_modules()


	def load_modules(self):
		"""
		Find and load all of our modules.
		"""
		print(f"Loading modules...")
		self._callables = {}
		self._hooks = []
		self.commands = {}
		self._times = {}
		self.url_callbacks = {}
		# ensure they're empty

		modules = loader.find_modules(self.config.homedir)
		loaded = 0
		failed = 0
		for name, path in modules.items():
			try:
				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'):
			self._callables[func.__name__] = func
			for cmd in func.commands:
				self.commands[cmd] = tools.Command(cmd)
				self.commands[cmd]._func_name = func.__name__
				self.commands[cmd].priv = func.priv
				self.commands[cmd].doc = func._docs
				if cmd in func.aliases:
					self.commands[cmd].canonical = False
				aliases = [a for a in func.commands if a != cmd]
				self.commands[cmd].aliases = aliases

		if func.hook:
			self._callables[func.__name__] = func
			self._hooks.append(func.__name__)

		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


	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'):
			self._callables.pop(func.__name__)
			for command in func.commands:
				self.commands.pop(command)

		if func.hook:
			self._callables.pop(func.__name__)
			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)


	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()


	## 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.
		"""
		func_names = []		
		if message.startswith(self.prefix) and message != self.prefix:
			command, _, _ = message.partition(" ")
			command = command.replace(self.prefix, "", 1)
			func_name = self.commands.get(command)._func_name
			if not func_name:
				return
			func_names.append(func_name)

		func_names += self._hooks

		for func_name in func_names:
			func = self._callables[func_name]
			trigger = Trigger(user, channel, message, "PRIVMSG", self.config)
			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."""
		print(f"Joined {channel}")
		if channel not in self.channels:
			self.channels[channel] = tools.Channel(channel)


	def left(self, channel):
		"""Called when the leaves a channel."""
		print(f"Parted {channel}")
		self.channels.pop(channel)


	def modeChanged(self, user, channel, add_mode, modes, args):
		"""Called when users or channel's modes are changed."""
		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]
				op_level = tools.op_level[mode]
				if add_mode:
					self.channels[channel].privileges[nick] += op_level
				else:
					self.channels[channel].privileges[nick] -= op_level

			else: # user mode, non-op_level mode
				continue


	def signedOn(self):
		"""Called when the bot successfully connects to the server."""
		print(f"Signed on as {self.nickname}")
		for channel in self.config.core.channels.split(","):
			self.join(channel)


	def kickedFrom(self, channel, kicker, message):
		"""Called when the bot is kicked from a channel."""
		self.channels.pop(channel)


	## 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."""
		if user not in self.users:
			self.users[user] = tools.User(user)
		self.channels[channel].add_user(self.users[user])


	def userLeft(self, user, channel):
		"""Called when the bot sees another user join a channel."""
		self.channels[channel].remove_user(user)


	def userQuit(self, user, quitMessage):
		"""Called when the bot sees another user disconnect from the network."""
		channels = list(self.users[user].channels.keys())
		for channel in channels:
			# self.users[user].channels[channel].remove_user
			# channel.remove_user(user)
			self.channels[channel].remove_user(user)
		self.users.pop(user)


	def userKicked(self, kickee, channel, kicker, message):
		"""
		Called when the bot sees another user getting kicked from the channel.
		"""
		self.channels[channel].remove_user(kickee)


	def topicUpdated(self, user, channel, newTopic):
		"""Called when the bot sees a user update the channel's topic."""
		self.channels[channel].topic = newTopic


	def userRenamed(self, oldname, newname):
		"""Called wehn the bot sees a user change their nickname."""
		user = self.users.pop(oldname)
		self.users[newname] = user
		for key, channel in user.channels.items():
			channel.rename_user(oldname, newname)
		user.nick = newname


	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 = tools.op_level.get(nick[0], 0)
			if op_level > 0:
				nick = nick[1:]
			self.users[nick] = tools.User(nick)
			self.channels[channel].add_user(self.users[nick])
			self.channels[channel].privileges[nick] = op_level


	## User commands, from client->server

	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)


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):
	# black magic going on here
	protocol = property(lambda s: functools.partial(Fulvia, s.config))

	def __init__(self, config):
		self.config = config